普通Maven项目使用RabbitMQ
一、简单模式
一个服务者对应一个消费者
-
创建BaseProject父maven项目，在父项目底下创建maven模块producer
-
导入相关依赖
<!-- rabbitmq依赖 -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.9.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.3.2</version>
</dependency>
<!-- 整合到spring项目需要导入此依赖 -->
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>2.3.10</version>
</dependency>
- 在resource目录下创建log4j.properties
log4j.rootLogger=DEBUG,A1 log4j.logger.com.taotao = DEBUG
log4j.logger.org.mybatis = DEBUG
log4j.appender.A1=org.apache.log4j.ConsoleAppender
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
log4j.appender.A1.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c]-[%p] %m%n
- 创建RabbitMqUtil工具类
private static ConnectionFactory factory;
public static Connection getConnection() throws IOException, TimeoutException {
//创建mq的连接工厂对象
factory = new ConnectionFactory();
//设置连接rabbit的主机地址
factory.setHost("主机ip");
//设置连接端口
factory.setPort(5672);
//设置连接哪个虚拟主机
factory.setVirtualHost("host1");
//设置虚拟主机的用户名和密码
factory.setUsername("zhangsan");
factory.setPassword("123");
Connection connection = factory.newConnection();
return connection;
}
-
服务提供者类发送消息
向队列queue1发送消息，如果该消息没被消费者消费，消息是不会消失的。
public class SendMsg {
public static void main(String[] args) {
String msg = "hello RabbitMq123!";
Connection connection = null;
try {
//获取连接,相当于JDBC的获取数据库连接
connection = MQUtil.getConnection();
/**
* 相当于JDBC的statement
* 声明（创建）队列,如果存在就不创建,不存在就创建
* 参数1 队列名,
* 参数2 durable： 是否持久化, 默认是存放到内存中，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库
* 参数3 exclusive：是否排外的,当当前链接关闭时，是否删除这个队列
* 参数4 autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除（无论队列中是否有数据）
* 参数5 arguments： 参数
* channel.queueDeclare("queue1", false, false, true, null);
*/
Channel channel = connection.createChannel();
/**
* 四个参数：
* 参数1 ： 交换机名称，
* 参数2 ： 目标队列名称，
* 参数3 ： 设置当前这条消息的属性（比如设置超时时间）
* 参数4 ： 消息内容
*/
channel.basicPublish("","queue1",null,msg.getBytes());
System.out.println("发送--->" + msg);
//关闭
channel.close();
connection.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
- 创建服务消费者模块consumer
导包、log4j配置、MQUtil工具类，与上述一样。
消费消息，消息消费过后，队列中的消息会清空。
//接受消息
public class ReceiveMsg {
public static void main(String[] args) {
Connection connection = null;
//获取连接,相当于JDBC的获取数据库连接
try {
connection = MQUtil.getConnection();
Channel channel = connection.createChannel();
//声明要关注的队列
//channel.queueDeclare("queue1", false, false, false, null);
//DefaultConsumer类实现了Consumer接口，通过传入一个频道，
// 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
Consumer consumer = new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("消费者消费消息：'" + message + "'");
}
};
//自动回复队列应答 -- RabbitMQ中的消息确认机制
channel.basicConsume("queue1", true, consumer);
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
总结：发送消息：channel.basicPublish()
接收消息：channel.basicConsume()
二、工作模式（一对多）
一个服务者对应多个消费者
在简单模式的基础上多加一个消费者，即创建模块consumer2，所有配置信息与consumer1一致。这里需要注意的是，consumer1和consumer2接收的队列、producer发送的目标队列都应为同一队列。
然后启动consumer1和consumer2，再启动producer，可以实时的看到，发送的消息，由两个消费者随机消费。
三、订阅模式
一个服务者对应多个消费者，但是中间由交换机转发。服务者发送消息->交换机->同时发给多个消费者。
准备工作：在测试之前，先确保两个队列在同一个交换机底下（本次模拟交换机为exc1，队列为queue1和queue2）
在工作模式的基础上做小修改：
consumer1模块接收队列queue1的消息、consumer2模块接收队列queue2的消息。如下图：
注明消息生产者的交换机名称：
然后启动consumer1和consumer2，再启动producer，可以实时的看到，发送的消息，由两个消费者共同消费，服务者发送的任何消息，所有的消费者都能接收到。
四、路由模式
一个服务者对应多个消费者，中间由交换机转发，交换机有对应的key，指定到对应key的队列。服务者发送消息（指定key）->交换机（根据key转发）->由key发给对应的消费者。
准备工作：在测试之前，先确保两个队列在同一个交换机底下，并且每个队列都有对应的routing key（本次模拟交换机为exc1，队列为queue1和queue2，queue1的key为a，queue2的key为b）
在订阅模式的基础上做小修改：
producer发送消息时指定对应的key：文章来源：https://www.toymoban.com/news/detail-624628.html
//发送消息
public class SendMsg {
public static void main(String[] args) {
System.out.println("请输入消息：");
Scanner input = new Scanner(System.in);
String msg = input.nextLine();
// String msg = "hello RabbitMq123!";
Connection connection = null;
try {
//获取连接,相当于JDBC的获取数据库连接
connection = MQUtil.getConnection();
/**
* 相当于JDBC的statement
* 声明（创建）队列,如果存在就不创建,不存在就创建
* 参数1 队列名,
* 参数2 durable： 是否持久化, 默认是存放到内存中，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库
* 参数3 exclusive：是否排外的,当当前链接关闭时，是否删除这个队列
* 参数4 autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除（无论队列中是否有数据）
* 参数5 arguments： 参数
* channel.queueDeclare("queue1", false, false, true, null);
*/
Channel channel = connection.createChannel();
/**
* 四个参数：
* 参数1 ： 交换机名称，
* 参数2 ： 目标队列名称，
* 参数3 ： 设置当前这条消息的属性（比如设置超时时间）
* 参数4 ： 消息内容
*/
if (msg.startsWith("a")){
channel.basicPublish("exc1","a",null, msg.getBytes());
} else if (msg.startsWith("b")){
channel.basicPublish("exc1","b",null, msg.getBytes());
}
System.out.println("发送--->" + msg);
//关闭
channel.close();
connection.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
则消费者只接收到自己key的消息。文章来源地址https://www.toymoban.com/news/detail-624628.html
到了这里，关于普通Maven项目使用RabbitMQ的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
