RabbitMQ简介：
RabbitMQ是一个实现了AMQP（Advanced Message Queuing Protocol）高级消息队列协议的消息队列服务，用Erlang语言。是面向消息的中间件。
你可以把它想像成一个邮局：你把信件放入邮箱，邮递员就会把信件投递到你的收件人处。在这个比喻中，RabbitMQ是一个邮箱、邮局、邮递员。RabbitMQ和邮局的主要区别是，它处理的不是纸，而是接收、存储和发送二进制的数据——消息。
主要流程：生产者（Producer）与消费者（Consumer）和 RabbitMQ 服务（Broker）建立连接， 然后生产者发布消息（Message）同时需要携带交换机（Exchange） 名称以及路由规则（Routing Key），这样消息会到达指定的交换机，然后交换机根据路由规则匹配对应的 Binding，最终将消息发送到匹配的消息队列（Quene），最后 RabbitMQ 服务将队列中的消息投递给订阅了该队列的消费者（消费者也可以主动拉取消息）。
这里总结了rabbitmq的五种模式代码案例，全部手敲测试通过，简单记录
准备环节：
rabbitmq，创建用户，分配主机
创建工程并引入依赖
<dependencies>
<!-- rabbitmq 依赖-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.9.0</version>
</dependency>
</dependencies>
1. 简单模式：Hello_world
单个生产者生产消息，直接通过队列进行发送至单个消费者，消费者消费
生产者代码
package com.yutao.producer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*
* @author yt
* @create 2022/10/14 13:40
*/
public class Producer_HelloWorld{
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置参数
factory.setHost("192.168.149.129");
//端口
factory.setPort(5672);
//自己在网页上创建的虚拟机
factory.setVirtualHost("/itcast");
//账号密码
factory.setUsername("yutao");
factory.setPassword("yutao");
//创建连接
Connection connection = factory.newConnection();
//创建channel
Channel channel = connection.createChannel();
/**
* 创建队列Queue
* 参数：
* 1.queue：队列名称
* 2.durable：是否持久化，当mq重启之后是否还在
* 3.exclusive：是否独占，只能有一个消费者监听这个队列 当connection关闭时是否=删除队列
* 4.aotoDelete:是否自动删除，当没有接收端时是否自动删除 null
* 5.arguments：参数
*/
//如果没有hello_world的队列会自动创建，有就不会创建
channel.queueDeclare("hello_world",true,false,false,null);
//发送消息
/**
* 1.exchange:交换机名称。简单模式用默认的“”
* 2.routingket：路由名称
* 3.props：配置信息
* 4.body：字节数组 发送的数据
*/
String body = "你好呀，我是于涛";
channel.basicPublish("","hello_world",null,body.getBytes());
channel.close();
connection.close();
}
}
消费者代码
package com.yutao.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_HelloWorld{
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置参数
factory.setHost("192.168.149.129");
//端口
factory.setPort(5672);
//自己在网页上创建的虚拟机
factory.setVirtualHost("/itcast");
//账号密码
factory.setUsername("yutao");
factory.setPassword("yutao");
//创建连接
Connection connection = factory.newConnection();
//创建channel
Channel channel = connection.createChannel();
/**
* 创建队列Queue
* 参数：
* 1.queue：队列名称
* 2.durable：是否持久化，当mq重启之后是否还在
* 3.exclusive：是否独占，只能有一个消费者监听这个队列 当connection关闭时是否=删除队列
* 4.aotoDelete:是否自动删除，当没有接收端时是否自动删除 null
* 5.arguments：参数
*/
//如果没有hello_world的队列会自动创建，有就不会创建
channel.queueDeclare("hello_world",true,false,false,null);
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel){
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @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("Exchange = " + envelope.getExchange());
System.out.println("RoutingKey = " + envelope.getRoutingKey());
System.out.println("properties = " + properties);
System.out.println("body = " + new String(body));
}
};
channel.basicConsume("hello_world",true,consumer);
}
}
抽取工具类
package com.yutao.producer.util;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author yt
* @create 2022/10/14 14:30
*/
public class RabbitUtils {
public static ConnectionFactory connectionFactory;
static {
connectionFactory = new ConnectionFactory();
//设置参数
connectionFactory.setHost("192.168.149.129");
//端口
connectionFactory.setPort(5672);
//自己在网页上创建的虚拟机
connectionFactory.setVirtualHost("/itcast");
//账号密码
connectionFactory.setUsername("yutao");
connectionFactory.setPassword("yutao");
}
public static Connection getConnection(){
try {
return connectionFactory.newConnection();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
return null;
}
public static void closeConnectionAndchannel(Channel channel,Connection connection){
try {
if (channel!=null){
channel.close();
}
if (connection!=null){
connection.close();
}
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
2.工作模式：work_queues
多个消费者争抢一条消息消费，一条消息只能被一个消费者消费，相当于对简单模式的拓展
让多个消费者绑定到一个队列共同消费队列中的消息。默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息
生产者代码：发送10条消息
package com.yutao.producer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
* 工作队列发送消息，一个发送者发送消息 n个消费者监听 只能有一个消费者消费
*
* @author yt
* @create 2022/10/14 13:40
*/
public class Producer_WorkQueues {
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
/**
* 创建队列Queue
* 参数：
* 1.queue：队列名称
* 2.durable：是否持久化，当mq重启之后是否还在
* 3.exclusive：是否独占，只能有一个消费者监听这个队列 当connection关闭时是否=删除队列
* 4.aotoDelete:是否自动删除，当没有接收端时是否自动删除 null
* 5.arguments：参数
*/
//如果没有hello_world的队列会自动创建，有就不会创建
channel.queueDeclare("work_queues", true, false, false, null);
//发送消息
/**
* 1.exchange:交换机名称。简单模式用默认的“”
* 2.routingket：路由名称
* 3.props：配置信息
* 4.body：字节数组 发送的数据
*/
for (int i = 0; i < 10; i++) {
String body = i + "你好呀，我是于涛";
channel.basicPublish("", "work_queues", null, body.getBytes());
}
channel.close();
connection.close();
}
}
创建两个消费者（代码相同）：
package com.yutao.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_WorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
/**
* 创建队列Queue
* 参数：
* 1.queue：队列名称
* 2.durable：是否持久化，当mq重启之后是否还在
* 3.exclusive：是否独占，只能有一个消费者监听这个队列 当connection关闭时是否=删除队列
* 4.aotoDelete:是否自动删除，当没有接收端时是否自动删除 null
* 5.arguments：参数
*/
//如果没有hello_world的队列会自动创建，有就不会创建
channel.queueDeclare("work_queues",true,false,false,null);
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel){
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @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("Exchange = " + envelope.getExchange());
System.out.println("RoutingKey = " + envelope.getRoutingKey());
System.out.println("properties = " + properties);*/
System.out.println("body = " + new String(body));
}
};
channel.basicConsume("work_queues",true,consumer);
}
}
消费结果
消费者1：
消费者2
3.订阅模式：pub/sub
生产者将消息发送给交换机，交换机将消息分散到队列中，然后消费者在对应的队列中消费
交换机(Exchange)有三种：
Fanout 广播：将消息交给所有绑定到交换机的队列
Diect 定向：把消息交给符合指定的key的队列
Topic 通配符：把消息交给符合routing pattern(路由模式的队列)
交换机只负责转发消息，不具备存储消息的能力，没有队列绑定交换机，或者没有路由规则的队列，消息将丢失！！！
生产者代码：
创建交换机
创建两个队列
将两个队列绑定至交换机上
发送消息到交换机上
关闭资源
package com.yutao.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
* 生产者将消息发送给交换机，交换机将消息分散到队列中，然后消费者在对应的队列中消费
*
* 创建交换机
* 创建两个队列
* 将两个队列绑定至交换机上
* 发送消息到交换机上
* 关闭资源
*
* @author yt
* @create 2022/10/14 13:40
*/
public class Producer_PubSub {
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接
Connection connection = RabbitUtils.getConnection();
//创建channel 信道
Channel channel = connection.createChannel();
//创建交换机
/**
* String exchange,交换机名称
* BuiltinExchangeType type, 交换机类型 @See BuiltinExchangeType 定向 广播 通配符 参数匹配
* boolean durable, 是否持久化
* boolean autoDelete, 是否自动删除
* internal 内部使用 一般是false
* Map<String, Object> arguments 参数列表
*/
String ExchangeName = "test_fanout";
channel.exchangeDeclare(ExchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
//创建队列
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//绑定队列和交换机
/**
* String queue, 队列名称
* String exchange, 交换机名称
* String routingKey 路由key
*/
channel.queueBind(queue1Name,ExchangeName,"");
channel.queueBind(queue2Name,ExchangeName,"");
//发送消息
String body = "日志信息：张三调用了findAll方法 日志级别为：info";
channel.basicPublish(ExchangeName,"",null,body.getBytes());
//释放资源
RabbitUtils.closeConnectionAndchannel(channel,connection);
}
}
两个消费者分别监听两个队列即可
消费者一：接收消息保存至数据库
package com.yutao.consumer;
import com.rabbitmq.client.*;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
* 接收队列一消息
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_PubSub1 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
String queue1Name = "test_fanout_queue1";
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel) {
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @param properties 配置信息
* @param body 数据
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body = " + new String(body));
System.out.println("将日志信息保存至数据库");
}
};
//接收队列一
channel.basicConsume(queue1Name, true, consumer);
}
}
消费者二：接收消息打印至控制台
package com.yutao.consumer;
import com.rabbitmq.client.*;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
* 接收队列二消息
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_PubSub2 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
String queue2Name = "test_fanout_queue2";
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel){
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @param properties 配置信息
* @param body 数据
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body = " + new String(body));
System.out.println("将日志信息打印至控制台");
}
};
//接收队列二
channel.basicConsume(queue2Name,true,consumer);
}
}
4. 路由模式：Routing
生产者将消息发送至交换机，不同级别的日志走不同的队列，区分不同的消息，消费者通过订阅不同的队列来处理自己的逻辑
生产者代码：
队列一绑定error级别的日志
队列二绑定info，error，warning级别的日志
队列一只能接收到error的消息
队列二能收到info，error，warning的消息
package com.yutao.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*
* 队列一绑定error级别的日志
* 队列二绑定info，error，warning级别的日志
*
* 队列一只能接收到error的消息
* 队列二能收到info，error，warning的消息
* @author yt
* @create 2022/10/14 13:40
*/
public class Producer_Routing {
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接
Connection connection = RabbitUtils.getConnection();
//创建channel 信道
Channel channel = connection.createChannel();
//创建交换机
/**
* String exchange,交换机名称
* BuiltinExchangeType type, 交换机类型 @See BuiltinExchangeType 定向 广播 通配符 参数匹配
* boolean durable, 是否持久化
* boolean autoDelete, 是否自动删除
* internal 内部使用 一般是false
* Map<String, Object> arguments 参数列表
*/
String ExchangeName = "test_direct";
channel.exchangeDeclare(ExchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
//创建队列
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//绑定队列和交换机
/**
* String queue, 队列名称
* String exchange, 交换机名称
* String routingKey 路由key
*/
//队列一绑定error级别的日志
channel.queueBind(queue1Name,ExchangeName,"error");
//队列二绑定info，error，warning级别的日志
channel.queueBind(queue2Name,ExchangeName,"info");
channel.queueBind(queue2Name,ExchangeName,"error");
channel.queueBind(queue2Name,ExchangeName,"warning");
//发送消息
String body = "日志信息：张三调用了findAll方法 日志级别为：info";
//参数二指定绑定的队列
channel.basicPublish(ExchangeName,"info",null,body.getBytes());
//释放资源
RabbitUtils.closeConnectionAndchannel(channel,connection);
}
}
消费队列一（error）
package com.yutao.consumer;
import com.rabbitmq.client.*;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
* 接收队列一消息
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_Routing1 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
String queue1Name = "test_direct_queue1";
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel) {
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @param properties 配置信息
* @param body 数据
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body = " + new String(body));
System.out.println("将日志信息保存至数据库");
}
};
//接收队列一
channel.basicConsume(queue1Name, true, consumer);
}
}
无法收到info的消息 ，因为绑定的是error
消费者二（info，error，warning）
package com.yutao.consumer;
import com.rabbitmq.client.*;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
* 接收队列一消息
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_Routing2 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
String queue2Name = "test_direct_queue2";
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel) {
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @param properties 配置信息
* @param body 数据
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body = " + new String(body));
System.out.println("将日志信息打印至控制台");
}
};
//接收队列一
channel.basicConsume(queue2Name, true, consumer);
}
}
可以收到
当生产者发送error级别的消息时
队列一能收到
队列二能收到
Routing模式要求队列在绑定交换机时需要指定routing key,消费会转发到符合routing key的队列
5.通配符模式：Topics
生产者生产消息发送至交换机，交换机通过通配符将消息发送给能匹配上的队列，再由订阅队列的消费者进行消费
生产者代码：
package com.yutao.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 发送消息
*根据通配符发送指定队列
* @author yt
* @create 2022/10/14 13:40
*/
public class Producer_Topics {
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接
Connection connection = RabbitUtils.getConnection();
//创建channel 信道
Channel channel = connection.createChannel();
//创建交换机
/**
* String exchange,交换机名称
* BuiltinExchangeType type, 交换机类型 @See BuiltinExchangeType 定向 广播 通配符 参数匹配
* boolean durable, 是否持久化
* boolean autoDelete, 是否自动删除
* internal 内部使用 一般是false
* Map<String, Object> arguments 参数列表
*/
String ExchangeName = "test_topic";
channel.exchangeDeclare(ExchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
//创建队列
String queue1Name = "test_topic_queue1";
String queue2Name = "test_topic_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//绑定队列和交换机
/**
* String queue, 队列名称
* String exchange, 交换机名称
* String routingKey 路由key
*/
//队列一绑定所有以error为结尾的日志，和绑定order后面所有的日志
channel.queueBind(queue1Name,ExchangeName,"#.error");
channel.queueBind(queue1Name,ExchangeName,"order.*");
//队列二绑定所有级别的日志
channel.queueBind(queue2Name,ExchangeName,"*.*");
//发送消息
String body = "日志信息：张三调用了delete方法 日志级别为：order.info";
//参数二指定绑定的队列
channel.basicPublish(ExchangeName,"order.info",null,body.getBytes());
//释放资源
RabbitUtils.closeConnectionAndchannel(channel,connection);
}
}
消费者一（队列：test_topic_queue1）
package com.yutao.consumer;
import com.rabbitmq.client.*;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
* 接收队列一消息
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_Topic1 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
String queue1Name = "test_topic_queue1";
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel) {
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @param properties 配置信息
* @param body 数据
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body = " + new String(body));
System.out.println("将日志信息保存至数据库");
}
};
//接收队列一
channel.basicConsume(queue1Name, true, consumer);
}
}
消费者二（队列：test_topic_queue2）
package com.yutao.consumer;
import com.rabbitmq.client.*;
import util.RabbitUtils;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者
* 接收队列一消息
*
* @author yt
* @create 2022/10/14 13:39
*/
public class Consumer_Topic2 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
//创建channel
Channel channel = connection.createChannel();
String queue2Name = "test_topic_queue2";
//接收消息
/**
* 1.queue:队列名称
* 2.aotoACK：是否自动确认
* 3.callback：回调对象
*/
Consumer consumer = new DefaultConsumer(channel) {
/**
* 回调方法。当收到消息之后会自动执行该方法
* @param consumerTag 标识
* @param envelope 获取一些信息，交换机，路由Key
* @param properties 配置信息
* @param body 数据
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("body = " + new String(body));
System.out.println("将日志信息保存至数据库");
}
};
//接收队列一
channel.basicConsume(queue2Name, true, consumer);
}
}
通配符和定向相比，通配符更加灵活！
SpringBoot整合RabbitMq
生产者：
创建工程，导入依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.3.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>2.3.10.RELEASE</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
</dependency>
</dependencies>
在application.yml中导入配置信息
#基本信息
spring:
rabbitmq:
#ip
host: 192.168.149.129
#端口
port: 5672
username: guest
password: guest
#虚拟机
virtual-host: /
创建rabbitmq的配置，创建交换机，并绑定队列，这里演示的是通配符的配置
package com.yutao.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 主要是创建一个交换机，创建一个队列，将二者绑定上即可，这里只简单列出方法，后续可以在此基础上进行拓展
* @author yt
* @create 2022/10/14 16:44
*/
@Configuration
public class RabbitMQConfig {
//交换机名称
public static final String EXCANGE_NAME = "boot_topic_echange";
//队列名称
public static final String QUEUE_NAME = "boot_queue";
//交换机
@Bean("bootExchange")
public Exchange bootExchange() {
//创建通配符交换机
return ExchangeBuilder.topicExchange(EXCANGE_NAME).durable(true).build();
}
//队列
@Bean("bootQueue")
public Queue bootQueue() {
return QueueBuilder.durable(QUEUE_NAME).build();
}
//绑定关系
/**
* 那个队列和那个交换机进行绑定.最后with的是routing Key 这里展示的是通配符的配置
*/
@Bean
public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
}
}
创建启动类
package com.yutao;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author yt
* @create 2022/10/14 16:43
*/
@SpringBootApplication
public class ProducerAppliction {
public static void main(String[] args) {
SpringApplication.run(ProducerAppliction.class,args);
}
}
创建测试类，导入rabbitTemplate，进行消息发送
convertAndSend参数分别是：交换机名称，routingkey，发送的消息
package com.yutao;
import com.yutao.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
* @author yt
* @create 2022/10/14 16:55
*/
@SpringBootTest
@RunWith(SpringRunner.class)
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSend() {
//convertAndSend参数分别是：交换机名称，routingkey，发送的消息
String msg = "springboot 整合 rabbitmq";
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCANGE_NAME, "boot.haha", msg);
}
}
消息发送成功！
消费者：
创建工程，引入依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.3.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<version>2.3.10.RELEASE</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.2</version>
<scope>test</scope>
</dependency>
</dependencies>
编写application.yml文件
#基本信息
spring:
rabbitmq:
#ip
host: 192.168.149.129
#端口
port: 5672
username: guest
password: guest
#虚拟机
virtual-host: /
创建启动类
package com.yutao;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author yt
* @create 2022/10/14 17:13
*/
@SpringBootApplication
public class ConsumerAppliction {
public static void main(String[] args) {
SpringApplication.run(ConsumerAppliction.class,args);
}
}
创建监听队列的方法，注意要注入spring容器
package com.yutao.rabbitmq.consumer;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @author yt
* @create 2022/10/14 17:15
*/
@Component
public class RabbitMQListener {
@RabbitListener(queues = "boot_queue")
public void ListenerQueue(Message message) {
System.out.println("message = " + message);
System.out.println("收到消息为：" + new String(message.getBody()));
}
}
启动之后即可收到消息
消息丢失：
在异常情况下，比如交换机挂了，消费者挂了，都会导致消息丢失的情况
1.生产者到rabbitmq消息丢失，这个可以用手动ack确认来解决
2.消息到达rabbitmq中丢失，这个可以用rabbit中的持久化来解决
3.rabbitmq到消费者过程丢失，不能判断是否被消费，这个也可以用回调的方式来解决
消息生产者到rabbitMq手动ACK确认
在原有的生产者代码的application.yml文件中增加如下配置
spring:
rabbitmq:
#ip
host: 120.27.224.146 #mq服务器的ip地址
port: 5678
virtual-host: /sbzq
username: kaying
password: hs3PGVzZD1rmrdSx
connection-timeout: 15000
//发送消息到交换机后会触发回调方法
publisher-confirm-type: correlated
添加回调方法即可显示成功与失败
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* 实现 RabbitTemplate.ConfirmCallback 接口 实现里面的回调方法
* 将 rabbitTemplate 用init方法注入到 ConfirmCallback中即可使用 @PostConstruct注解是将rabbitTemplate注入
* @author yt
* @create 2022/12/29 15:09
*/
@Component
public class Callback implements RabbitTemplate.ConfirmCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
//注入
@PostConstruct
public void init(){
rabbitTemplate.setConfirmCallback(this);
}
/**
*
* @param correlationData 保存了回调信息的id及相关信息（可以自己将发送的消息放到里面）
* @param ack 交换机是否收到消息true=收到了，false=没收到
* @param cause 失败原因，成功=null，失败会有失败原因，可以作为日志保存
*/
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) { // 消息投递到broker 的状态，true表示成功
System.out.println("消息发送到Broker成功！");
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
} else { // 发送异常
System.out.println("发送失败");
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
}
}
}
结果：
如果交换机到队列的时候找不到队列，将会删除消息造成消息丢失
配置文件中增加
#基本信息
spring:
rabbitmq:
#ip
host: 120.27.224.146 #mq服务器的ip地址
port: 5678
virtual-host: /sbzq
username: kaying
password: hs3PGVzZD1rmrdSx
connection-timeout: 15000
publisher-confirm-type: correlated
//回退消息
publisher-returns: true
实现回退接口并注入
package com.yutao.rabbitmq.config;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* 实现 RabbitTemplate.ConfirmCallback 接口 实现里面的回调方法
* 将 rabbitTemplate 用init方法注入到 ConfirmCallback中即可使用 @PostConstruct注解是将rabbitTemplate注入
* @author yt
* @create 2022/12/29 15:09
*/
@Component
public class Callback implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
//注入
@PostConstruct
public void init(){
//注入回调
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
/**
*
* @param correlationData 保存了回调信息的id及相关信息（可以自己将发送的消息放到里面）
* @param ack 交换机是否收到消息true=收到了，false=没收到
* @param cause 失败原因，成功=null，失败会有失败原因，可以作为日志保存
*/
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) { // 消息投递到broker 的状态，true表示成功
System.out.println("消息发送到Broker成功！");
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
} else { // 发送异常
System.out.println("发送失败");
System.out.println("correlationData = " + correlationData);
System.out.println("ack = " + ack);
System.out.println("cause = " + cause);
}
}
//在消息不可达消费者时将会退给生产者，只有不可达目的地时才会回退
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
System.out.println("returnedMessage = " + returnedMessage);
}
}
结果
本文转载自: https://blog.csdn.net/m0_58709145/article/details/127319766
版权归原作者 于京京9909 所有， 如有侵权，请联系我们删除。
版权归原作者 于京京9909 所有， 如有侵权，请联系我们删除。