中山手机网站建设费用,wordpress添加分类筛选手表,做百度移动网站优,深圳网站建设分期付1.Direct
直连(路由)交换机,生产者将消息发送到交换机#xff0c;并指定消息的Routing Key#xff08;路由键#xff09;。交换机会将Routing Key与队列绑定进行匹配#xff0c;如果匹配成功#xff0c;则将该消息路由到对应的队列中。如果没有匹配成功#xff0c;该消息…1.Direct
直连(路由)交换机,生产者将消息发送到交换机并指定消息的Routing Key路由键。交换机会将Routing Key与队列绑定进行匹配如果匹配成功则将该消息路由到对应的队列中。如果没有匹配成功该消息将被丢弃或返回给生产者。在Direct模式中每个消息只能被一个消费者接收。 通过使用Exchange和Routing Key来进行消息传输Direct模式实现了消息的有选择性地路由提高了消息传输的效率减少了系统负载. 如上图中的routingKey为error绑定队列disk,routingKey为info或warning绑定队列console。
2.生产者
package com.hong.rabbitmq7;import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;/*** Description: 直连模式消息发送者* Author: hong* Date: 2024-01-15 22:24* Version: 1.0**/
public class DirectSend {public static final String EXCHANGE_NAME direct_logs;public static void main(String[] args) throws Exception {Channel channel RabbitMQUtil.getChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);MapString,String map new HashMap();map.put(info,我是info信息);map.put(debug,我是debug信息);map.put(warning,我是warning信息);map.put(error,我是error信息);for(Map.EntryString,String bindingKeys : map.entrySet()){String bindingKey bindingKeys.getKey();String message bindingKeys.getValue();channel.basicPublish(EXCHANGE_NAME,bindingKey,null,message.getBytes(StandardCharsets.UTF_8));System.out.println(消息发送完成------ message);}}
}3.消费者1-disk只接受error消息
package com.hong.rabbitmq7;import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.nio.charset.StandardCharsets;/*** Description: 直连模式消息接收者1-disk接收error消息* Author: hong* Date: 2024-01-15 20:22* Version: 1.0**/
public class Receiver1 {public static final String EXCHANGE_NAME direct_logs;public static void main(String[] args) throws Exception {Channel channel RabbitMQUtil.getChannel();/**声明交换机*第1个参数:交换机名称*第2个参数:交换机类型*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);//声明队列String queueName disk;channel.queueDeclare(queueName,false,false,false,null);/** 绑定队列与交换机* 第1个参数:队列名称* 第2个参数:交换机名称* 第3个参数:routingKey*/channel.queueBind(queueName,EXCHANGE_NAME,error);DeliverCallback deliverCallback (comsumerTag, message) - {System.out.println(disk中的: new String(message.getBody(), StandardCharsets.UTF_8));};CancelCallback cancelCallback var - {};channel.basicConsume(queueName,true,deliverCallback,cancelCallback);}
}4.消费2-console接收info和warning消息
package com.hong.rabbitmq7;import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.nio.charset.StandardCharsets;/*** Description: 直连模式消息接收者2-console接收info和warning消息* Author: hong* Date: 2024-01-15 20:22* Version: 1.0**/
public class Receiver2 {public static final String EXCHANGE_NAME direct_logs;public static void main(String[] args) throws Exception {Channel channel RabbitMQUtil.getChannel();/**声明交换机*第1个参数:交换机名称*第2个参数:交换机类型*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);//声明队列String queueName console;channel.queueDeclare(queueName,false,false,false,null);/** 绑定队列与交换机* 第1个参数:队列名称* 第2个参数:交换机名称* 第3个参数:routingKey*/channel.queueBind(queueName,EXCHANGE_NAME,info);channel.queueBind(queueName,EXCHANGE_NAME,warning);DeliverCallback deliverCallback (comsumerTag, message) - {System.out.println(console中的: new String(message.getBody(), StandardCharsets.UTF_8));};CancelCallback cancelCallback var - {};channel.basicConsume(queueName,true,deliverCallback,cancelCallback);}
}
