当前位置：首页 > news >正文

rabbitMQ的direct模式的生产者与消费者使用案例

news 2025/8/14 6:10:11

在这里插入图片描述

消费者C1的RoutingKey 规则按照info warn 两种RoutingKey匹配绑定队列console

package com.esint.rabbitmq.work03;import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;/*** 消费者01的消息接受*/
public class ReceiveLog01 {public static final String EXCHANGE_NAME = "logs";public static void main(String[] args) throws Exception {//获取信道Channel channel = RabbitMQUtils.getChannel();//声明一个交换机channel.exchangeDeclare(EXCHANGE_NAME,"fanout");/***声明一个队列  队列名字随机* 消费者断开与队列的链接后 队列自动删除*/String queue = channel.queueDeclare().getQueue();/*** 绑定交换机与队列* 1.队列名字*/channel.queueBind(queue,EXCHANGE_NAME,"");System.out.println("wait print message...");DeliverCallback deliverCallback = (var,var2)->{System.out.println("01:"+new String(var2.getBody()));};channel.basicConsume(queue,true,deliverCallback, var->{});}
}

消费者C2的RoutingKey 规则按照error RoutingKey匹配绑定队列disk

package com.esint.rabbitmq.work03;import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;/*** 消费者02的消息接受*/
public class ReceiveLog02 {public static final String EXCHANGE_NAME = "logs";public static void main(String[] args) throws Exception {//获取信道Channel channel = RabbitMQUtils.getChannel();//声明一个交换机channel.exchangeDeclare(EXCHANGE_NAME,"fanout");/***声明一个队列  队列名字随机* 消费者断开与队列的链接后 队列自动删除*/String queue = channel.queueDeclare().getQueue();/*** 绑定交换机与队列* 1.队列名字*/channel.queueBind(queue,EXCHANGE_NAME,"");System.out.println("wait print message...");DeliverCallback deliverCallback = (var,var2)->{System.out.println("02:"+new String(var2.getBody(),"UTF-8"));};channel.basicConsume(queue,true,deliverCallback, var->{});}
}

生产者发送基于下面调试程序改变参数RoutingKey 切换info warn error 可以观察两个生产者的不同数据接收

package com.esint.rabbitmq.work03;import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;import java.util.Scanner;public class EmitLog {public static final String EXCHANGE_NAME = "logs";public static void main(String[] args) throws Exception {Channel channel = RabbitMQUtils.getChannel();//声明交换机channel.exchangeDeclare(EXCHANGE_NAME,"fanout");Scanner scanner = new Scanner(System.in);while(scanner.hasNext()){String mes = scanner.next();channel.basicPublish(EXCHANGE_NAME,"",null,mes.getBytes("UTF-8"));System.out.println("生产者发出消息："+mes.toString());}}
}