RabbitMQ Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了。下面分别以实例的方式对这几种exchange进行讲解。
direct
首先我们以路由的方式对消息进行过滤，代码如下：
生产者
1 public class RoutingSendDirect {
2
3 private static final String EXCHANGE_NAME = "direct_test";
4
5 private static final String[] routingKeys = new String[]{"info" ,"warning", "error"};
6
7 public static void main(String[] args) throws IOException, TimeoutException {
8 ConnectionFactory connectionFactory = new ConnectionFactory();
9 connectionFactory.setHost("localhost");
10 Connection connection = connectionFactory.newConnection();
11 Channel channel = connection.createChannel();
12 channel.exchangeDeclare(EXCHANGE_NAME,"direct");
13 for(String key : routingKeys){
14 String message = "RoutingSendDirect Send the message level:" + key;
15 channel.basicPublish(EXCHANGE_NAME,key,null,message.getBytes());
16 System.out.println("RoutingSendDirect Send"+key +"‘:‘" + message);
17 }
18 channel.close();
19 connection.close();
20 }
21 }
消费者
1 public class ReceiveDirect1 {
2 // 交换器名称
3 private static final String EXCHANGE_NAME = "direct_test";
4 // 路由关键字
5 private static final String[] routingKeys = new String[]{"info" ,"warning"};
6
7 public static void main(String[] args) throws IOException, TimeoutException {
8 ConnectionFactory connectionFactory = new ConnectionFactory();
9 connectionFactory.setHost("localhost");
10 Connection connection = connectionFactory.newConnection();
11 Channel channel = connection.createChannel();
12 channel.exchangeDeclare(EXCHANGE_NAME,"direct");
13 //获取匿名队列名称
14 String queueName=channel.queueDeclare().getQueue();
15 for(String key : routingKeys){
16 channel.queueBind(queueName,EXCHANGE_NAME,key);
17 System.out.println("ReceiveDirect1 exchange:"+EXCHANGE_NAME+"," +
18 " queue:"+queueName+", BindRoutingKey:" + key);
19 }
20
21 System.out.println("ReceiveDirect1 Waiting for messages");
22 Consumer consumer = new DefaultConsumer(channel){
23 @Override
24 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25 String msg = new String(body,"UTF-8");
26 System.out.println("ReceiveDirect1 Received ‘" + envelope.getRoutingKey() + "‘:‘" + msg + "‘");
27 }
28 };
29
30 channel.basicConsume(queueName, true, consumer);
31 }
32 }
1 public class ReceiveDirect2 {
2 // 交换器名称
3 private static final String EXCHANGE_NAME = "direct_test";
4 // 路由关键字
5 private static final String[] routingKeys = new String[]{"error"};
6
7 public static void main(String[] args) throws IOException, TimeoutException {
8 ConnectionFactory connectionFactory = new ConnectionFactory();
9 connectionFactory.setHost("localhost");
10 Connection connection = connectionFactory.newConnection();
11 Channel channel = connection.createChannel();
12 channel.exchangeDeclare(EXCHANGE_NAME,"direct");
13 //获取匿名队列名称
14 String queueName=channel.queueDeclare().getQueue();
15 for(String key : routingKeys){
16 channel.queueBind(queueName,EXCHANGE_NAME,key);
17 System.out.println("ReceiveDirect2 exchange:"+EXCHANGE_NAME+"," +
18 " queue:"+queueName+", BindRoutingKey:" + key);
19 }
20
21 System.out.println("ReceiveDirect2 Waiting for messages");
22 Consumer consumer = new DefaultConsumer(channel){
23 @Override
24 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25 String msg = new String(body,"UTF-8");
26 System.out.println("ReceiveDirect2 Received ‘" + envelope.getRoutingKey() + "‘:‘" + msg + "‘");
27 }
28 };
29
30 channel.basicConsume(queueName, true, consumer);
31 }
32 }
运行结果如下：
1 RoutingSendDirect Sendinfo‘:‘RoutingSendDirect Send the message level:info 2 RoutingSendDirect Sendwarning‘:‘RoutingSendDirect Send the message level:warning 3 RoutingSendDirect Senderror‘:‘RoutingSendDirect Send the message level:error 4 5 ReceiveDirect1 exchange:direct_test, queue:amq.gen-HsUrzbjzto-K7HeigXSEfQ, BindRoutingKey:info 6 ReceiveDirect1 exchange:direct_test, queue:amq.gen-HsUrzbjzto-K7HeigXSEfQ, BindRoutingKey:warning 7 ReceiveDirect1 Waiting for messages 8 ReceiveDirect1 Received ‘info‘:‘RoutingSendDirect Send the message level:info‘ 9 ReceiveDirect1 Received ‘warning‘:‘RoutingSendDirect Send the message level:warning‘ 10 11 ReceiveDirect2 exchange:direct_test, queue:amq.gen-i3NY12l3DqWjGapaBOCdwQ, BindRoutingKey:error 12 ReceiveDirect2 Waiting for messages 13 ReceiveDirect2 Received ‘error‘:‘RoutingSendDirect Send the message level:error‘
fanout
fanout和别的ＭＱ的发布/订阅模式类似，实例如下：
生产者
1 public class Pub {
2 private static final String EXCHANGE_NAME = "logs";
3 public static void main(String[] args) throws IOException, TimeoutException {
4 ConnectionFactory factory=new ConnectionFactory();
5 factory.setHost("localhost");
6 Connection connection=factory.newConnection();
7 Channel channel=connection.createChannel();
8 //fanout表示分发，所有的消费者得到同样的队列信息
9 channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
10 //分发信息
11 for (int i=0;i<5;i++){
12 String message="Hello World"+i;
13 channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
14 System.out.println("Pub Sent ‘" + message + "‘");
15 }
16 channel.close();
17 connection.close();
18 }
19 }
消费者
public class Sub {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//产生一个随机的队列名称
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");//对队列进行绑定
System.out.println("Sub Waiting for messages");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Sub Received ‘" + message + "‘");
}
};
channel.basicConsume(queueName, true, consumer);//队列会自动删除
}
}
Topics
这种应该属于模糊匹配，* ：可以替代一个词，#：可以替代0或者更多的词，现在我们继续看看代码来理解
生产者
1 public class TopicSend {
2 private static final String EXCHANGE_NAME = "topic_logs";
3
4 public static void main(String[] args) throws IOException, TimeoutException {
5 Connection connection = null;
6 Channel channel = null;
7 try{
8 ConnectionFactory factory=new ConnectionFactory();
9 factory.setHost("localhost");
10 connection=factory.newConnection();
11 channel=connection.createChannel();
12
13 //声明一个匹配模式的交换机
14 channel.exchangeDeclare(EXCHANGE_NAME,"topic");
15 //待发送的消息
16 String[] routingKeys=new String[]{
17 "quick.orange.rabbit",
18 "lazy.orange.elephant",
19 "quick.orange.fox",
20 "lazy.brown.fox",
21 "quick.brown.fox",
22 "quick.orange.male.rabbit",
23 "lazy.orange.male.rabbit"
24 };
25 //发送消息
26 for(String severity :routingKeys){
27 String message = "From "+severity+" routingKey‘ s message!";
28 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
29 System.out.println("TopicSend Sent ‘" + severity + "‘:‘" + message + "‘");
30 }
31 }catch (Exception e){
32 e.printStackTrace();
33 if (connection!=null){
34 channel.close();
35 connection.close();
36 }
37 }finally {
38 if (connection!=null){
39 channel.close();
40 connection.close();
41 }
42 }
43 }
44 }
消费者
1 public class ReceiveLogsTopic1 {
2 private static final String EXCHANGE_NAME = "topic_logs";
3
4 public static void main(String[] args) throws IOException, TimeoutException {
5 ConnectionFactory factory = new ConnectionFactory();
6 factory.setHost("localhost");
7 Connection connection = factory.newConnection();
8 Channel channel = connection.createChannel();
9
10 //声明一个匹配模式的交换机
11 channel.exchangeDeclare(EXCHANGE_NAME, "topic");
12 String queueName = channel.queueDeclare().getQueue();
13 //路由关键字
14 String[] routingKeys = new String[]{"*.orange.*"};
15 //绑定路由
16 for (String routingKey : routingKeys) {
17 channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
18 System.out.println("ReceiveLogsTopic1 exchange:" + EXCHANGE_NAME + ", queue:" + queueName + ", BindRoutingKey:" + routingKey);
19 }
20 System.out.println("ReceiveLogsTopic1 Waiting for messages");
21
22 Consumer consumer = new DefaultConsumer(channel) {
23 @Override
24 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25 String message = new String(body, "UTF-8");
26 System.out.println("ReceiveLogsTopic1 Received ‘" + envelope.getRoutingKey() + "‘:‘" + message + "‘");
27 }
28 };
29 channel.basicConsume(queueName, true, consumer);
30 }
31 }
1 public class ReceiveLogsTopic2 {
2 private static final String EXCHANGE_NAME = "topic_logs";
3
4 public static void main(String[] argv) throws IOException, TimeoutException {
5 ConnectionFactory factory = new ConnectionFactory();
6 factory.setHost("localhost");
7 Connection connection = factory.newConnection();
8 Channel channel = connection.createChannel();
9 // 声明一个匹配模式的交换器
10 channel.exchangeDeclare(EXCHANGE_NAME, "topic");
11 String queueName = channel.queueDeclare().getQueue();
12 // 路由关键字
13 String[] routingKeys = new String[]{"*.*.rabbit", "lazy.#"};
14 // 绑定路由关键字
15 for (String bindingKey : routingKeys) {
16 channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
17 System.out.println("ReceiveLogsTopic2 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + bindingKey);
18 }
19
20 System.out.println("ReceiveLogsTopic2 Waiting for messages");
21
22 Consumer consumer = new DefaultConsumer(channel) {
23 @Override
24 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws UnsupportedEncodingException {
25 String message = new String(body, "UTF-8");
26 System.out.println("ReceiveLogsTopic2 Received ‘" + envelope.getRoutingKey() + "‘:‘" + message + "‘");
27 }
28 };
29 channel.basicConsume(queueName, true, consumer);
30 }
31 }
原文地址：https://www.cnblogs.com/senlinyang/p/8440627.html
时间： 2024-11-05 19:38:32