在虚拟机上安装Erlang的GCC环境，装erlong，然后安装rabbitmq
参考：安装说明链接
安装web端面板
创建交换机
先学习一下工作模式（详细介绍可见官网）
上代码
1.Hello Word模式
写在测试类中：
Providucer
@Test
void contextLoads()throws Exception {
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 1.queue 队列名
* 2.durable 是否持久化
* 3.exclusive (1)是否独占站，只能有一个消费的监听者 (2)当Connection关闭时是否删除队列
* 4.autoDelete 是否自动删除，当没有consumer会自动删除掉
* 5.arguments：
* */
channel.queueDeclare("peng",true,false,false,null);
//6.发送消息
//basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
/**
* 1.exchange:交换机名称
* 2.routingKey:路由名称
* 3.props:配置信息
* 4.body:发送的消息数据
*/
String body="第一个消息";
channel.basicPublish("","peng",null,body.getBytes());
//7.释放资源
channel.close();
connection.close();
}
Consumer
@Test
void contextLoads()throws Exception {
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 1.queue 队列名
* 2.durable 是否持久化
* 3.exclusive (1)是否独占站，只能有一个消费的监听者 (2)当Connection关闭时是否删除队列
* 4.autoDelete 是否自动删除，当没有consumer会自动删除掉
* 5.arguments：
* */
channel.queueDeclare("peng",true,false,false,null);
//basicConsume(String queue, DeliverCallback deliverCallback, CancelCallback cancelCallback)
/**
* 1.queue:队列名
* 2.deliverCallback：是否自动确认收到
* 3.cancelCallback：回调对象
*/
Consumer consumer= new DefaultConsumer(channel){
/**
* 1.consumerTag:
* 2.envelope:
* 3.properties:
* 4.body:
* @param consumerTag
* @param envelope
* @param properties
* @param body
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("consumerTag"+consumerTag);
System.out.println("envelope"+envelope.getExchange());
System.out.println("properties"+envelope.getRoutingKey());
System.out.println("properties"+properties);
System.out.println("body"+new String(body));
}
};
channel.basicConsume("peng",true,consumer);
}
2.Work Queues模式
生产者生产，两个消费者循环消费
P：
package com.providucer.factory;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* @ClassName: Providucerfactory
* @author: 鹏
* @date: 2023/7/4 14:42
*/
public class ProvideFactory {
public static void main(String[] args) throws Exception{
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 1.queue 队列名
* 2.durable 是否持久化
* 3.exclusive (1)是否独占站，只能有一个消费的监听者 (2)当Connection关闭时是否删除队列
* 4.autoDelete 是否自动删除，当没有consumer会自动删除掉
* 5.arguments：
* */
channel.queueDeclare("pengwork",true,false,false,null);
//6.发送消息
//basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
/**
* 1.exchange:交换机名称
* 2.routingKey:路由名称
* 3.props:配置信息
* 4.body:发送的消息数据
*/
for (int i = 1; i <= 10; i++) {
String body="第"+i+"个消息";
channel.basicPublish("","pengwork",null,body.getBytes());
}
//7.释放资源
channel.close();
connection.close();
}
}
C1：
package com.consumer.factory;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
* @ClassName: ConsumerFactory
* @author: 鹏
* @date: 2023/7/4 14:44
*/
public class ConsumerFactory {
public static void main(String[] args)throws Exception {
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 1.queue 队列名
* 2.durable 是否持久化
* 3.exclusive (1)是否独占站，只能有一个消费的监听者 (2)当Connection关闭时是否删除队列
* 4.autoDelete 是否自动删除，当没有consumer会自动删除掉
* 5.arguments：
* */
channel.queueDeclare("pengwork",true,false,false,null);
//basicConsume(String queue, DeliverCallback deliverCallback, CancelCallback cancelCallback)
/**
* 1.queue:队列名
* 2.deliverCallback：是否自动确认收到
* 3.cancelCallback：回调对象
*/
Consumer consumer= new DefaultConsumer(channel){
/**
* 1.consumerTag:
* 2.envelope:
* 3.properties:
* 4.body:
* @param consumerTag
* @param envelope
* @param properties
* @param body
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
/* System.out.println("consumerTag"+consumerTag);
System.out.println("envelope"+envelope.getExchange());
System.out.println("properties"+envelope.getRoutingKey());
System.out.println("properties"+properties);*/
System.out.println("body"+new String(body));
}
};
channel.basicConsume("pengwork",true,consumer);
}
}
C2：
package com.consumer.factory;
import com.rabbitmq.client.*;
import java.io.IOException;
/**
* @ClassName: ConsumerFactory
* @author: 鹏
* @date: 2023/7/4 14:44
*/
public class ConsumerFactory1 {
public static void main(String[] args)throws Exception {
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 1.queue 队列名
* 2.durable 是否持久化
* 3.exclusive (1)是否独占站，只能有一个消费的监听者 (2)当Connection关闭时是否删除队列
* 4.autoDelete 是否自动删除，当没有consumer会自动删除掉
* 5.arguments：
* */
channel.queueDeclare("pengwork",true,false,false,null);
//basicConsume(String queue, DeliverCallback deliverCallback, CancelCallback cancelCallback)
/**
* 1.queue:队列名
* 2.deliverCallback：是否自动确认收到
* 3.cancelCallback：回调对象
*/
Consumer consumer= new DefaultConsumer(channel){
/**
* 1.consumerTag:
* 2.envelope:
* 3.properties:
* 4.body:
* @param consumerTag
* @param envelope
* @param properties
* @param body
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
/* System.out.println("consumerTag"+consumerTag);
System.out.println("envelope"+envelope.getExchange());
System.out.println("properties"+envelope.getRoutingKey());
System.out.println("properties"+properties);*/
System.out.println("body"+new String(body));
}
};
channel.basicConsume("pengwork",true,consumer);
}
}
消费结果：
3.Publish/Subscribe订阅模式
消费着只需要绑定相应的队列，生产者需要创建交换机
public class PubFactory {
public static void main(String[] args) throws Exception{
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建交换机
//exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
String exchange="test_fanout";
channel.exchangeDeclare(exchange, BuiltinExchangeType.FANOUT,
true,false,false,null);
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
//1.exchange:交换机名称
//2.type:交换机类型
/**
* DIRECT("direct"),定向
* FANOUT("fanout"),扇形（广播）
* TOPIC("topic"),通配符方式
* HEADERS("headers")参数匹配
*/
//3.durable:是否持久化
//4.autoDelete:是福哦自动删除
//5.internal: 内部使用一般用false
//6.arguments: 参数
//channel.exchangeDeclare();
//6.创建队列
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7.绑定交换机与队列
/**
* 1.queue:队列名称
* 2.exchange：交换机名称
* 3.routingKey:路由键，绑定规则
* 如果交换机的类型为fanout，routingKey设置为""
*/
channel.queueBind(queue1Name,exchange,"");
channel.queueBind(queue2Name,exchange,"");
String body="日志信息：接收成功";
channel.basicPublish(exchange,"",null,body.getBytes());
//8.释放资源
channel.close();
connection.close();
}
}
4.Routing路由模式
路由模式相当于增加一层限制，只有通过相应的限制交换机才能将消息发布到对应的队列，也就是在发布的时候路由参数数设置值，且交换机类型必须为directchannel.basicPublish(exchange,"error",null,body.getBytes());此处限制队列路由为error的可以发送文章来源：https://www.toymoban.com/news/detail-543242.html
public class RoutingFactory {
public static void main(String[] args) throws Exception{
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建交换机
//exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
String exchange="test_direct";
channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT,
true,false,false,null);
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
//1.exchange:交换机名称
//2.type:交换机类型
/**
* DIRECT("direct"),定向
* FANOUT("fanout"),扇形（广播）
* TOPIC("topic"),通配符方式
* HEADERS("headers")参数匹配
*/
//3.durable:是否持久化
//4.autoDelete:是福哦自动删除
//5.internal: 内部使用一般用false
//6.arguments: 参数
//channel.exchangeDeclare();
//6.创建队列
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7.绑定交换机与队列
/**
* 1.queue:队列名称
* 2.exchange：交换机名称
* 3.routingKey:路由键，绑定规则
* 如果交换机的类型为fanout，routingKey设置为""
*/
channel.queueBind(queue1Name,exchange,"error");
channel.queueBind(queue2Name,exchange,"info");
channel.queueBind(queue2Name,exchange,"error");
channel.queueBind(queue2Name,exchange,"warming");
String body="日志信息：接收成功";
channel.basicPublish(exchange,"error",null,body.getBytes());
//8.释放资源
channel.close();
connection.close();
}
}
5. Topics模式
相对于routing在队列增加了匹配规则，让交换机发送与队列接受更加灵活*匹配一个单词，#匹配多个单词 channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true,false,false,null);
设置为BuiltinExchangeType.TOPIC
文章来源地址https://www.toymoban.com/news/detail-543242.html
public class TopicsFactory {
public static void main(String[] args) throws Exception{
//1.创建链接
ConnectionFactory factory = new ConnectionFactory();
//2。设置参数
factory.setHost("192.168.63.130");
factory.setPort(5672);
factory.setVirtualHost("/peng");
factory.setUsername("peng");
factory.setPassword("peng");
//3.创建链接Connection
Connection connection = factory.newConnection();
//4.创建Channel
Channel channel = connection.createChannel();
//5.创建交换机
//exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
String exchange="test_topic";
channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC,
true,false,false,null);
String queue1Name = "test_topic_queue1";
String queue2Name = "test_topic_queue2";
//1.exchange:交换机名称
//2.type:交换机类型
/**
* DIRECT("direct"),定向
* FANOUT("fanout"),扇形（广播）
* TOPIC("topic"),通配符方式
* HEADERS("headers")参数匹配
*/
//3.durable:是否持久化
//4.autoDelete:是福哦自动删除
//5.internal: 内部使用一般用false
//6.arguments: 参数
//channel.exchangeDeclare();
//6.创建队列
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7.绑定交换机与队列
/**
* 1.queue:队列名称
* 2.exchange：交换机名称
* 3.routingKey:路由键，绑定规则
* 如果交换机的类型为fanout，routingKey设置为""
*/
channel.queueBind(queue1Name,exchange,"*.*");
channel.queueBind(queue2Name,exchange,"*.one");
channel.queueBind(queue2Name,exchange,"*.two");
channel.queueBind(queue2Name,exchange,"ok.*");
String body="日志信息：接收成功";
channel.basicPublish(exchange,"error",null,body.getBytes());
channel.basicPublish(exchange,"123.one",null,body.getBytes());
channel.basicPublish(exchange,"123.two",null,body.getBytes());
//8.释放资源
channel.close();
connection.close();
}
}
到了这里，关于RabbitMQ-基础学习的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
