本帖最后由 执迷不悟 于 2019-3-14 15:54 编辑
rabbitmq入门学习四
1. 路由模型(Routing)路由模式： 1、每个消费者监听自己的队列，并且设置routingkey。 2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。 1.1 生产者代码 以下生产者代码中交换机绑定了两个队列，每个队列分别有两个不同的routingkey [Java] 纯文本查看 复制代码 public class RoutingProducer {
//队列名称
private static final String QUEUE_INFORM_1 = "queue_inform_1";
private static final String QUEUE_INFORM_2 = "queue_inform_2";
private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";
private static final String ROUTINGKEY_1="inform_1";
private static final String ROUTINGKEY_2="inform_2";
public static void main(String[] args) {
Connection connection = null;
Channel channel = null;
try {
//建立新连接
connection = ConnectionUtil.getConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
channel = connection.createChannel();
//声明队列，如果队列在mq 中没有则要创建
channel.queueDeclare(QUEUE_INFORM_1,true,false,false,null);
channel.queueDeclare(QUEUE_INFORM_2,true,false,false,null);
//声明一个交换机，交换机类型为direct
channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
//进行交换机和队列绑定
channel.queueBind(QUEUE_INFORM_1,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_1);
channel.queueBind(QUEUE_INFORM_1,EXCHANGE_ROUTING_INFORM,"inform");
channel.queueBind(QUEUE_INFORM_2,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_2);
channel.queueBind(QUEUE_INFORM_2,EXCHANGE_ROUTING_INFORM,"inform");
//发送消息
for(int i=0;i<5;i++){
//发送消息的时候指定routingKey
String message = "send inform message to user";
//向routingkey是inform的队列中发送消息
//channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform",null,message.getBytes());
//向routingkey是inform_1的队列中发送消息
//channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform_1",null,message.getBytes());
//向routingkey是inform_2的队列中发送消息
channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform_2",null,message.getBytes());
System.out.println("send to mq "+message);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//关闭连接
//先关闭通道
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
[Java] 纯文本查看 复制代码 public class RoutingComsumer1 {
//队列名称
private static final String QUEUE_INFORM_1 = "queue_inform_1";
private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";
private static final String ROUTINGKEY_1="inform_1";
public static void main(String[] args) throws Exception {
//建立新连接
Connection connection = ConnectionUtil.getConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_INFORM_1,true,false,false,null);
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
//进行交换机和队列绑定
channel.queueBind(QUEUE_INFORM_1, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_1);
//实现消费方法
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
String exchange = envelope.getExchange();
//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String message= new String(body,"utf-8");
System.out.println("【消费者1】:"+message);
}
};
//监听队列
channel.basicConsume(QUEUE_INFORM_1,true,defaultConsumer);
}
}
1.3 消费者代码2
[Java] 纯文本查看 复制代码 public class RoutingComsumer2 {
//队列名称
private static final String QUEUE_INFORM_2 = "queue_inform_2";
private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";
private static final String ROUTINGKEY_2="inform_2";
public static void main(String[] args) throws Exception {
//建立新连接
Connection connection = ConnectionUtil.getConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_INFORM_2,true,false,false,null);
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
//进行交换机和队列绑定
channel.queueBind(QUEUE_INFORM_2, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_2);
//实现消费方法
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//交换机
String exchange = envelope.getExchange();
//消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String message= new String(body,"utf-8");
System.out.println("【消费者2】:"+message);
}
};
//监听队列
channel.basicConsume(QUEUE_INFORM_2,true,defaultConsumer);
}
}
1.4 测试 运行生产者代码测试，我们发现交换机绑定的队列中routingkey有值了，分别为： Inform，inform_1，inform，inform_2 运行生产者进行效果测试： 1. 如果我们执行的是如下的消息发送，两个消费者都会接收到消息。 channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform",null,message.getBytes()) 效果截图： 两个消费者都可以接收到消息
2. 将消息发送到inform_1,那么只有消费者1可以接收到消息 annel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform_1",null,message.getBytes())
3. 将消息发送到inform_2,那么只有消费者2可以接收到消息 annel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform_2",null,message.getBytes()) 问题： 如果你在消费者代码中没有绑定routingKey，那么在生产者发送的到队列的消息都可以接收，不管生产者指定的routingkey是什么。
|
收藏
淘帖
踩
