前言
- 学习自bili尚硅谷-RabbitMQ
发布确认
- 之前的消息应答，队列持久化是为了保证 -> 消息从rabbitmq队列到消费者的过程中不会丢失；消息持久化则是为了保证 -> 消息从生产者到队列的过程中不会丢失，但也不能完全保证，因此有了发布确认策略，即当消息到达队列后，会通知生产者以确认
- 单个发布确认：发布一个确认一个，效率低
- 批量发布确认：批量发布，批量确认，若消息丢失，则不知道是那条消息丢失了
- 异步发布确认业务逻辑：
- 生产者向队列发送消息，生产者只管发送消息，发送消息前会有一个监听器，监听发布确认的消息，和未发布确认的消息、
- 发布消息前会将所有消息记录到一个并发链路队列，当监听器监听到消息发布确认后，就将并发链路队列中的消息删除，剩余则时未发布确认的消息
点击查看详细代码
public static void publishMessageAsync() throws Exception {
try (Channel channel = RabbitMqUtils.getChannel()) {
String queueName = UUID.randomUUID().toString(); // 随机生成队列名称
channel.queueDeclare(queueName, false, false, false, null);
// 开启发布确认
channel.confirmSelect();
// 新建一个并发链路队列
ConcurrentSkipListMap<Long, String> outstandingConfirms = new
ConcurrentSkipListMap<>();
// 确认收到消息的一个回调
ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
/**
* 返回批量确认消息或单个确认消息时，清除
*/
if (multiple) {
ConcurrentNavigableMap<Long, String> confirmed =
outstandingConfirms.headMap(sequenceNumber, true);
confirmed.clear();
}else{
outstandingConfirms.remove(sequenceNumber);
}
};
// 未确认收到消息的回调，打印出队列没有收到的消息
ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
String message = outstandingConfirms.get(sequenceNumber);
System.out.println("发布的消息"+message+"未被确认，序列号"+sequenceNumber);
};
// 添加一个异步确认的监听器，参数是两个回调函数，队列收到消息的回调以及未收到消息的回调
channel.addConfirmListener(ackCallback, nackCallback);
long begin = System.currentTimeMillis(); // 获取发送消息的开始时间
for (int i = 0; i < MESSAGE_COUNT; i++){ // 发送消息
String message = "消息" + i;
// 发送消息前，将所有发送的消息记录到一个并发链路队列
outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
channel.basicPublish("", queueName, null, message.getBytes()); // 发送消息
}
long end = System.currentTimeMillis();
System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) +"ms");
}
}
交换机
- 生产者发消息给交换机，交换机通过routingKey绑定不同的队列，队列再将消息发送给不同的消费者
- 交换机类型：设置为空，则表示使用默认交换机；直接(direct)；主题(topic)；标题(headers)；扇出(fanout)
扇出
即像广播一样，同时向多个消费者发送同样的消息
- 编写两个类作为消费者，队列名称是随机生成的
点击查看详细代码
public class ReceiveLogs01 {
private static final String EXCHANGE_NAME = "logs"; // 交换机名称
public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); // 获取信道
channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 声明一个交换机
// 生成一个临时的队列 队列的名称是随机的，当消费者断开和该队列的连接时，队列自动删除
String queueName = channel.queueDeclare().getQueue();
// 队列绑定到交换机，参数：队列、交换机、routingKey可为null
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println("等待接收消息,把接收到的消息打印在屏幕........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> { // 接收成功的方法回调
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("控制台打印接收到的消息"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); // 接收消息
}
}
- 编写生产者发送消息：
点击查看详细代码
public class EmitLog {
private static final String EXCHANGE_NAME = "logs"; // 交换机名称
public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitUtils.getChannel()) {
// 声明一个交换机：名称、类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
Scanner sc = new Scanner(System.in);
System.out.println("请输入信息");
while (sc.hasNext()) {
String message = sc.nextLine();
// 交换机、路由key、消息持久化、消息体
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); // 发送消息
System.out.println("生产者发出消息" + message);
}
}
}
}
- 测试：生产者发送消息，两个消费者都能收到消息
- 生产者到交换机、交换机到两个消费者，绑定的路由key是相同的，所以生产者发送的消息，两个消费者都能收到
直接
- 新建一个类作为消费者：交换机 -> 路由key(error) -> 队列disk -> 消费者1
点击查看详细代码
public class ReceiveLogsDirect01 {
private static final String EXCHANGE_NAME = "direct_logs"; // 交换机名称
public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel(); // 获取信道
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); // 声明交换机
String queueName = "disk"; // 声明队列
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "error"); // 绑定路由key为error
System.out.println("等待接收消息........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> { // 接收成功的回调
String message = new String(delivery.getBody(), "UTF-8");
message="接收绑定键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message;
File file = new File("C:\\work\\rabbitmq_info.txt");
FileUtils.writeStringToFile(file,message,"UTF-8");
System.out.println("错误日志已经接收");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
});
}
}
- 再新建一个类作为消费者，交换机和队列间绑定了两条路由key，即交换机可通过任意一条路由将消息传递给队列：交换机 -> 路由key(info、warning) -> 队列console -> 消费者2
- 新建类作为生产者，使用不同的路由key，则进入不同的队列：生产者 -> 路由key -> 队列 -> 消费者
点击查看详细代码
public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitUtils.getChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); // 声明一个交换机，并指定类型
//创建多个 bindingKey
Map<String, String> bindingKeyMap = new HashMap<>();
bindingKeyMap.put("info","普通 info 信息");
bindingKeyMap.put("warning","警告 warning 信息");
bindingKeyMap.put("error","错误 error 信息");
bindingKeyMap.put("debug","调试 debug 信息"); // 该条路由key不存在，模拟消息丢失
for (Map.Entry<String, String> bindingKeyEntry: bindingKeyMap.entrySet()){
String bindingKey = bindingKeyEntry.getKey();
String message = bindingKeyEntry.getValue();
// 发送消息：交换机、路由key、消息是否持久化、消息体
channel.basicPublish(EXCHANGE_NAME,bindingKey, null,message.getBytes("UTF-8"));
System.out.println("生产者发出消息:" + message);
}
}
}
主题
- 扇出、直接模式的交换机都存在缺陷，扇出无法像指定的单个队列发送消息，直接模式无法同时向所有队列发送消息
主题模式则可以向指定的单个队列发送消息，也可以向所有队列发送消息
- 新建一个类作为消费者：交换机 -> 路由key -> 队列Q1 -> 消费者1
点击查看详细代码
public class ReceiveLogsTopic01 {
private static final String EXCHANGE_NAME = "topic_logs"; // 交换机名称
public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 声明一个交换机，指定类型为：主题
String queueName="Q1"; // 队列Q1
channel.queueDeclare(queueName, false, false, false, null);
channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*"); // 绑定的留有key：*.orange.*
System.out.println("等待接收消息........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" 接 收 队 列 :"+queueName+" 绑 定 键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
});
}
}
- 再新建一个类作为消费者：交换机 -> 路由key -> 队列Q1 -> 消费者1
点击查看详细代码
public class ReceiveLogsTopic02 {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws Exception {
Channel channel = RabbitUtils.getChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName="Q2";
channel.queueDeclare(queueName, false, false, false, null);
// 交换机和队列Q2之间绑定了两条路由key
channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#");
System.out.println("等待接收消息........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" 接 收 队 列 :"+queueName+" 绑 定 键:"+delivery.getEnvelope().getRoutingKey()+",消息:"+message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
});
}
}
- 新建类作为生产者，使用不同的路由key，则进入不同的队列：生产者 -> 路由key -> 队列 -> 消费者
点击查看详细代码
public class EmitLogTopic {
private static final String EXCHANGE_NAME = "topic_logs"; // 交换机名称
public static void main(String[] argv) throws Exception {
try (Channel channel = RabbitUtils.getChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 声明交换机，指定类型
Map<String, String> bindingKeyMap = new HashMap<>();
bindingKeyMap.put("quick.orange.rabbit","被队列 Q1Q2 接收到");
bindingKeyMap.put("lazy.orange.elephant","被队列 Q1Q2 接收到");
bindingKeyMap.put("quick.orange.fox","被队列 Q1 接收到");
bindingKeyMap.put("lazy.brown.fox","被队列 Q2 接收到");
bindingKeyMap.put("lazy.pink.rabbit","虽然满足两个绑定但只被队列 Q2 接收一次");
bindingKeyMap.put("quick.brown.fox","不匹配任何绑定不会被任何队列接收到会被丢弃");
bindingKeyMap.put("quick.orange.male.rabbit","是四个单词不匹配任何绑定会被丢弃");
bindingKeyMap.put("lazy.orange.male.rabbit","是四个单词但匹配 Q2");
for (Map.Entry<String, String> bindingKeyEntry: bindingKeyMap.entrySet()){
String bindingKey = bindingKeyEntry.getKey();
String message = bindingKeyEntry.getValue();
channel.basicPublish(EXCHANGE_NAME,bindingKey, null,
message.getBytes("UTF-8"));
System.out.println("生产者发出消息" + message);
}
}
}
}
星号表示一个单词，井号表示零个或多个单词；因此使用类似模糊查询的方式，生产者便可匹配需要的消费者