ToB企服应用市场:ToB评测及商务社交产业平台

标题: RabbitMQ工作模式详解，以及Java实现 [打印本页]

---

作者: 立山    时间: 2024-9-6 05:28
标题: RabbitMQ工作模式详解，以及Java实现
RabbitMQ共有六种工作模式：简单模式（Simple）、工作队列模式（Work Queue）、发布订阅模式（Publish/Subscribe）、路由模式（Routing）、通配符模式（Topics）、长途调用模式（RPC，不常用，不做讲讲授明）
   一、简单模式（Simple）

特点：①一个生产者对应一个消耗者，通过队列举行消息传递。
           ②该模式使用direct交换机，direct交换机是RabbitMQ默认交换机。
图解：

二、工作队列模式（Work Queue）

与简单模式相比，工作队列模式(Work Queue)多了一些消耗者，该模式也使用direct交换机，应用于处置惩罚消息较多的环境。特点如下：
①一个队列对应多个消耗者。
②一条消息只会被一个消耗者消耗。
③消息队列默认采用轮询的方式将消息均匀发送给消耗者。
图解：

三、发布订阅模式（Publish/Subscribe）

在开发过程中，有一些消息需要不同消耗者举行不同的处置惩罚
特点：
①生产者将消息发送给交换机，交换机将消息转发到绑定此交换机的每个队列中。
②工作队列模式的交换机只能将消息发送给一个队列，发布订阅模式的交换性能将消息发送给多个队列。发布订阅模式使用fanout交换机。
图解

四、路由模式（Routing）

使用发布订阅模式时，所有消息都会发送到绑定的队列中，但很多时间，不是所有消息都无差异的发布到所有队列中。特点：
①每个队列绑定路由关键字RoutingKey
②生产者将带有RoutingKey的消息发送给交换机，交换机根据RoutingKey转发到指定队列。路由模式使用direct交换机。
图解：

五、 通配符模式（Topics）


通配符模式(Topic)是在路由模式的基础上，给队列绑定带通配符的路由关键字，只要消息的RoutingKey能实现通配符匹配，就会将消息转发到该队列。通配符模式比路由模式更灵活，使用topic交换机。
通配符规则：
①消息设置RoutingKey时，RoutingKey由多个单词构成，中央以.分割。
②队列设置RoutingKey时，#可以匹配任意多个单词，*可以匹配任意一个单词。
六、Java实现五种模式

---

   简单模式生产者

1. package com.tmh.mq.simple;
3. import com.rabbitmq.client.Channel;
4. import com.rabbitmq.client.Connection;
5. import com.rabbitmq.client.ConnectionFactory;
7. import java.io.IOException;
8. import java.util.concurrent.TimeoutException;
10. /**
11. * rabbitmq工作模式：简单模式，消息生产者
12. */
13. public class Producer {
14.     public static void main(String[] args) throws IOException, TimeoutException {
15.         //1、创建连接工厂 指定IP地址、端口、连接的用户名密码、连接的虚拟主机
16.         ConnectionFactory connectionFactory = new ConnectionFactory();
17.         connectionFactory.setHost("192.168.66.8");
18.         connectionFactory.setPort(5672);
19.         connectionFactory.setUsername("tangminghao");
20.         connectionFactory.setPassword("tangminghao");
21.         connectionFactory.setVirtualHost("/");
22.         //2、创建连接
23.          Connection connection = connectionFactory.newConnection();
24.         //3、创建信道
25.          Channel channel = connection.createChannel();
26.         //4、创建队列
27.         /**
28.          * 参数一：队列名称
29.          * 参数二：是否持久化，true表示当MQ重启后队列还在
30.          * 参数三：是否私有化，false表示所有消费者都可以访问，true表示只有第一次拥有它的消费者才能访问
31.          * 参数四：是否自动删除
32.          * 参数五：其他参数
33.          */
34.         channel.queueDeclare("simplqueue", false,false,false,null);
35.         //5、发送消息
36.         String messges="hello simple queue";
37.         /**
38.          * 参数1：交换机名，""表示默认交换机
39.          * 参数2：路由键，简单模式就是队列名
40.          * 参数3：其他额外参数
41.          * 参数4：要传递的消息字节数组
42.          */
43.         channel.basicPublish("","simplqueue",null,messges.getBytes());
44.         //6、关闭信道和连接
45.         channel.close();
46.         connection.close();
47.         System.out.println("消息发送成功");
49.     }
50. }

复制代码

简单模式消耗者

1. package com.tmh.mq.simple;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. //simple队列消息消费者
9. public class Consumer {
10.     public static void main(String[] args) throws IOException, TimeoutException {
11.         //1、创建连接工厂 指定IP地址、端口、连接的用户名密码、连接的虚拟主机
12.         ConnectionFactory connectionFactory = new ConnectionFactory();
13.         connectionFactory.setHost("192.168.66.8");
14.         connectionFactory.setPort(5672);
15.         connectionFactory.setUsername("tangminghao");
16.         connectionFactory.setPassword("tangminghao");
17.         connectionFactory.setVirtualHost("/");
18.         //2、创建连接
19.         Connection connection = connectionFactory.newConnection();
20.         //3、创建信道
21.         Channel channel = connection.createChannel();
22.         //4、监听队列
23.         /**
24.          * 参数1：监听的队列名
25.          * 参数2：是否自动签收，如果设置为false，则需要手动确认消息已收到，否则MQ会一直发送消息
26.          * 参数3：Consumer的实现类，重写该类方法表示接受到消息后如何消费
27.          */
28.         channel.basicConsume("simplqueue",true,new DefaultConsumer(channel){
29.             @Override
30.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
31.                  String messge = new String(body, "UTF-8");
32.                 System.out.println("接收消息，消息为："+messge);
33.             }
34.         });
35.     }
36. }

复制代码

---

   工作队列模式生产者

1. package com.tmh.mq.work;
3. import com.rabbitmq.client.Channel;
4. import com.rabbitmq.client.Connection;
5. import com.rabbitmq.client.ConnectionFactory;
6. import com.rabbitmq.client.MessageProperties;
8. import java.io.IOException;
9. import java.util.concurrent.TimeoutException;
11. //工作队列模式
12. public class Producer {
13.     public static void main(String[] args) throws IOException, TimeoutException {
14.         //创建工厂连接
15.         ConnectionFactory connectionFactory=new ConnectionFactory();
16.         connectionFactory.setHost("192.168.66.8");
17.         connectionFactory.setPort(5672);
18.         connectionFactory.setUsername("tangminghao");
19.         connectionFactory.setPassword("tangminghao");
20.         connectionFactory.setVirtualHost("/");
21.         //建立连接
22.          Connection connection = connectionFactory.newConnection();
23.          //创建信道
24.          Channel channel = connection.createChannel();
25.          //创建队列，如果队列已经存在，则使用该队列
26.         channel.queueDeclare("workqueue",true,false,false,null);
27.         //发送大量消息
28.         for (int i = 0; i <100 ; i++) {
29.             channel.basicPublish("","workqueue",
30.                     MessageProperties.PERSISTENT_TEXT_PLAIN,
31.                     ("hello 这是今天的第"+(i+1)+"条消息").getBytes());
33.         }
34.         //关闭资源
35.         channel.close();
36.         connection.close();
37.     }
38. }

复制代码

 工作模式消耗者

1. package com.tmh.mq.work;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. public class Consumer1 {
9.     public static void main(String[] args) throws IOException, TimeoutException {
10.         //创建工厂连接
11.         ConnectionFactory connectionFactory=new ConnectionFactory();
12.         connectionFactory.setHost("192.168.66.8");
13.         connectionFactory.setPort(5672);
14.         connectionFactory.setUsername("tangminghao");
15.         connectionFactory.setPassword("tangminghao");
16.         connectionFactory.setVirtualHost("/");
17.         //建立连接
18.         Connection connection = connectionFactory.newConnection();
19.         //创建信道
20.         Channel channel = connection.createChannel();
21.         //监听队列
22.         channel.basicConsume("workqueue",true,new DefaultConsumer(channel){
23.             @Override
24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25.                  String s = new String(body, "UTF-8");
26.                 System.out.println("消费者1消费消息："+s);
27.             }
28.         });
29.     }
30. }
31. -----------------------------------------------------------------------------------------
32. public class Consumer2 {
33.     public static void main(String[] args) throws IOException, TimeoutException {
34.         //创建工厂连接
35.         ConnectionFactory connectionFactory=new ConnectionFactory();
36.         connectionFactory.setHost("192.168.66.8");
37.         connectionFactory.setPort(5672);
38.         connectionFactory.setUsername("tangminghao");
39.         connectionFactory.setPassword("tangminghao");
40.         connectionFactory.setVirtualHost("/");
41.         //建立连接
42.         Connection connection = connectionFactory.newConnection();
43.         //创建信道
44.         Channel channel = connection.createChannel();
45.         //监听队列
46.         channel.basicConsume("workqueue",true,new DefaultConsumer(channel){
47.             @Override
48.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
49.                  String s = new String(body, "UTF-8");
50.                 System.out.println("消费者2消费消息："+s);
51.             }
52.         });
53.     }
54. }
55. -----------------------------------------------------------------------------------------
56. public class Consumer3 {
57.     public static void main(String[] args) throws IOException, TimeoutException {
58.         //创建工厂连接
59.         ConnectionFactory connectionFactory=new ConnectionFactory();
60.         connectionFactory.setHost("192.168.66.8");
61.         connectionFactory.setPort(5672);
62.         connectionFactory.setUsername("tangminghao");
63.         connectionFactory.setPassword("tangminghao");
64.         connectionFactory.setVirtualHost("/");
65.         //建立连接
66.         Connection connection = connectionFactory.newConnection();
67.         //创建信道
68.         Channel channel = connection.createChannel();
69.         //监听队列
70.         channel.basicConsume("workqueue",true,new DefaultConsumer(channel){
71.             @Override
72.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
73.                  String s = new String(body, "UTF-8");
74.                 System.out.println("消费者1消费消息："+s);
75.             }
76.         });
77.     }
78. }

复制代码

---

   发布订阅模式生产者

1. package com.tmh.mq.publish;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.Connection;
6. import com.rabbitmq.client.ConnectionFactory;
8. import java.io.IOException;
9. import java.nio.charset.StandardCharsets;
10. import java.util.concurrent.TimeoutException;
12. public class Producer {
13.     public static void main(String[] args) throws IOException, TimeoutException {
14.         //1、创建连接工厂
15.         final ConnectionFactory connectionFactory = new ConnectionFactory();
16.         connectionFactory.setHost("192.168.66.8");
17.         connectionFactory.setPort(5672);
18.         connectionFactory.setUsername("tangminghao");
19.         connectionFactory.setPassword("tangminghao");
20.         connectionFactory.setVirtualHost("/");
21.         //2、建立连接
22.         final Connection connection = connectionFactory.newConnection();
23.         //3、建立信道
24.         final Channel channel = connection.createChannel();
25.         //4、创建交换机
26.         /**
27.          * 参数一：交换机名
28.          * 参数二：交换机类型
29.          * 参数三：交换机持久化
30.          */
31.         channel.exchangeDeclare("exchange_fanout", BuiltinExchangeType.FANOUT,true);
32.         //5、创建队列
33.         channel.queueDeclare("MAIL_QUEUE", true,false,false,null);
34.         channel.queueDeclare("MESSAGE_QUEUE", true,false,false,null);
35.         channel.queueDeclare("STATION_QUEUE", true,false,false,null);
36.         //6、队列绑定交换机
37.         /**
38.          * 参数一：队列名
39.          * 参数二：交换机名字
40.          * 参数三：路由关键字，发布订阅模式写“”即可
41.          */
42.         channel.queueBind("MAIL_QUEUE","exchange_fanout","");
43.         channel.queueBind("MESSAGE_QUEUE","exchange_fanout","");
44.         channel.queueBind("STATION_QUEUE","exchange_fanout","");
45.         //7、发送消息
46.         for (int i = 1; i <=10 ; i++) {
47.             channel.basicPublish("exchange_fanout","",null,
48.                     ("你好，发布订阅模式"+i).getBytes(StandardCharsets.UTF_8));
49.         }
50.         //8、关闭资源
51.         channel.close();
52.         connection.close();
53.     }
54. }

复制代码

 发布订阅模式消耗者

1. package com.tmh.mq.publish;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. public class ConsumerMail {
9.     public static void main(String[] args) throws IOException, TimeoutException {
10.         //1、创建连接工厂
11.         final ConnectionFactory connectionFactory = new ConnectionFactory();
12.         connectionFactory.setHost("192.168.66.8");
13.         connectionFactory.setPort(5672);
14.         connectionFactory.setUsername("tangminghao");
15.         connectionFactory.setPassword("tangminghao");
16.         connectionFactory.setVirtualHost("/");
17.         //2、建立连接
18.         final Connection connection = connectionFactory.newConnection();
19.         //3、建立信道
20.         final Channel channel = connection.createChannel();
21.         //4、监听队列
22.         channel.basicConsume("MAIL_QUEUE",true,new DefaultConsumer(channel){
23.             @Override
24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25.                 final String s = new String(body, "UTF-8");
26.                 System.out.println("邮件消息："+s);
27.             }
28.         });
30.     }
31. }
32. -----------------------------------------------------------------------------------------
33. public class ConsumerMessage {
34.     public static void main(String[] args) throws IOException, TimeoutException {
35.         //1、创建连接工厂
36.         final ConnectionFactory connectionFactory = new ConnectionFactory();
37.         connectionFactory.setHost("192.168.66.8");
38.         connectionFactory.setPort(5672);
39.         connectionFactory.setUsername("tangminghao");
40.         connectionFactory.setPassword("tangminghao");
41.         connectionFactory.setVirtualHost("/");
42.         //2、建立连接
43.         final Connection connection = connectionFactory.newConnection();
44.         //3、建立信道
45.         final Channel channel = connection.createChannel();
46.         //4、监听队列
47.         channel.basicConsume("MESSAGE_QUEUE",true,new DefaultConsumer(channel){
48.             @Override
49.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
50.                 final String s = new String(body, "UTF-8");
51.                 System.out.println("短信消息："+s);
52.             }
53.         });
55.     }
56. }
57. -----------------------------------------------------------------------------------------
58. public class ConsumerStation {
59.     public static void main(String[] args) throws IOException, TimeoutException {
60.         //1、创建连接工厂
61.         final ConnectionFactory connectionFactory = new ConnectionFactory();
62.         connectionFactory.setHost("192.168.66.8");
63.         connectionFactory.setPort(5672);
64.         connectionFactory.setUsername("tangminghao");
65.         connectionFactory.setPassword("tangminghao");
66.         connectionFactory.setVirtualHost("/");
67.         //2、建立连接
68.         final Connection connection = connectionFactory.newConnection();
69.         //3、建立信道
70.         final Channel channel = connection.createChannel();
71.         //4、监听队列
72.         channel.basicConsume("STATION_QUEUE",true,new DefaultConsumer(channel){
73.             @Override
74.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
75.                 final String s = new String(body, "UTF-8");
76.                 System.out.println("站内消息："+s);
77.             }
78.         });
80.     }
81. }

复制代码

---

   路由模式生产者

1. package com.tmh.mq.routing;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.Connection;
6. import com.rabbitmq.client.ConnectionFactory;
8. import java.io.IOException;
9. import java.nio.charset.StandardCharsets;
10. import java.util.concurrent.TimeoutException;
12. public class Producer {
13.     public static void main(String[] args) throws IOException, TimeoutException {
14.         //1、创建连接工厂
15.         final ConnectionFactory connectionFactory = new ConnectionFactory();
16.         connectionFactory.setHost("192.168.66.8");
17.         connectionFactory.setPort(5672);
18.         connectionFactory.setUsername("tangminghao");
19.         connectionFactory.setPassword("tangminghao");
20.         connectionFactory.setVirtualHost("/");
21.         //2、建立连接
22.         final Connection connection = connectionFactory.newConnection();
23.         //3、建立信道
24.         final Channel channel = connection.createChannel();
25.         //4、创建交换机
26.         /**
27.          * 参数一：交换机名
28.          * 参数二：交换机类型
29.          * 参数三：交换机持久化
30.          */
31.         channel.exchangeDeclare("exchange_routing", BuiltinExchangeType.DIRECT,true);
32.         //5、创建队列
33.         channel.queueDeclare("MAIL_QUEUE2", true,false,false,null);
34.         channel.queueDeclare("MESSAGE_QUEUE2", true,false,false,null);
35.         channel.queueDeclare("STATION_QUEUE2", true,false,false,null);
36.         //6、队列绑定交换机
37.         /**
38.          * 参数一：队列名
39.          * 参数二：交换机名字
40.          * 参数三：路由关键字，发布订阅模式写“”即可
41.          */
42.         channel.queueBind("MAIL_QUEUE2","exchange_routing","important");
43.         channel.queueBind("MESSAGE_QUEUE2","exchange_routing","important");
44.         channel.queueBind("STATION_QUEUE2","exchange_routing","important");
45.         channel.queueBind("STATION_QUEUE2","exchange_routing","normal");
46.         //7、发送消息
48.             channel.basicPublish("exchange_routing","important",null,
49.                     ("你好，路由模式，这是重要消息").getBytes(StandardCharsets.UTF_8));
51.             channel.basicPublish("exchange_routing","normal",null,
52.                     ("你好，路由模式，这是不太重要的消息").getBytes(StandardCharsets.UTF_8));
54.         //8、关闭资源
55.         channel.close();
56.         connection.close();
57.     }
58. }

复制代码

路由模式消耗者

1. package com.tmh.mq.routing;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. public class ConsumerMail {
9.     public static void main(String[] args) throws IOException, TimeoutException {
10.         //1、创建连接工厂
11.         final ConnectionFactory connectionFactory = new ConnectionFactory();
12.         connectionFactory.setHost("192.168.66.8");
13.         connectionFactory.setPort(5672);
14.         connectionFactory.setUsername("tangminghao");
15.         connectionFactory.setPassword("tangminghao");
16.         connectionFactory.setVirtualHost("/");
17.         //2、建立连接
18.         final Connection connection = connectionFactory.newConnection();
19.         //3、建立信道
20.         final Channel channel = connection.createChannel();
21.         //4、监听队列
22.         channel.basicConsume("MAIL_QUEUE2",true,new DefaultConsumer(channel){
23.             @Override
24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25.                 final String s = new String(body, "UTF-8");
26.                 System.out.println("邮件消息："+s);
27.             }
28.         });
30.     }
31. }
32. -----------------------------------------------------------------------------------------
33. public class ConsumerMessage {
34.     public static void main(String[] args) throws IOException, TimeoutException {
35.         //1、创建连接工厂
36.         final ConnectionFactory connectionFactory = new ConnectionFactory();
37.         connectionFactory.setHost("192.168.66.8");
38.         connectionFactory.setPort(5672);
39.         connectionFactory.setUsername("tangminghao");
40.         connectionFactory.setPassword("tangminghao");
41.         connectionFactory.setVirtualHost("/");
42.         //2、建立连接
43.         final Connection connection = connectionFactory.newConnection();
44.         //3、建立信道
45.         final Channel channel = connection.createChannel();
46.         //4、监听队列
47.         channel.basicConsume("MESSAGE_QUEUE2",true,new DefaultConsumer(channel){
48.             @Override
49.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
50.                 final String s = new String(body, "UTF-8");
51.                 System.out.println("短信消息："+s);
52.             }
53.         });
55.     }
56. }
57. -----------------------------------------------------------------------------------------
58. public class ConsumerStation {
59.     public static void main(String[] args) throws IOException, TimeoutException {
60.         //1、创建连接工厂
61.         final ConnectionFactory connectionFactory = new ConnectionFactory();
62.         connectionFactory.setHost("192.168.66.8");
63.         connectionFactory.setPort(5672);
64.         connectionFactory.setUsername("tangminghao");
65.         connectionFactory.setPassword("tangminghao");
66.         connectionFactory.setVirtualHost("/");
67.         //2、建立连接
68.         final Connection connection = connectionFactory.newConnection();
69.         //3、建立信道
70.         final Channel channel = connection.createChannel();
71.         //4、监听队列
72.         channel.basicConsume("STATION_QUEUE2",true,new DefaultConsumer(channel){
73.             @Override
74.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
75.                 final String s = new String(body, "UTF-8");
76.                 System.out.println("站内消息："+s);
77.             }
78.         });
81.     }
82. }

复制代码

---

   通配符模式 生产者

1. package com.tmh.mq.topic;
3. import com.rabbitmq.client.BuiltinExchangeType;
4. import com.rabbitmq.client.Channel;
5. import com.rabbitmq.client.Connection;
6. import com.rabbitmq.client.ConnectionFactory;
8. import java.io.IOException;
9. import java.nio.charset.StandardCharsets;
10. import java.util.concurrent.TimeoutException;
12. public class Producer {
13.     public static void main(String[] args) throws IOException, TimeoutException {
14.         //1、创建连接工厂
15.         final ConnectionFactory connectionFactory = new ConnectionFactory();
16.         connectionFactory.setHost("192.168.66.8");
17.         connectionFactory.setPort(5672);
18.         connectionFactory.setUsername("tangminghao");
19.         connectionFactory.setPassword("tangminghao");
20.         connectionFactory.setVirtualHost("/");
21.         //2、建立连接
22.         final Connection connection = connectionFactory.newConnection();
23.         //3、建立信道
24.         final Channel channel = connection.createChannel();
25.         //4、创建交换机
26.         /**
27.          * 参数一：交换机名
28.          * 参数二：交换机类型
29.          * 参数三：交换机持久化
30.          */
31.         channel.exchangeDeclare("exchange_topic", BuiltinExchangeType.TOPIC,true);
32.         //5、创建队列
33.         channel.queueDeclare("MAIL_QUEUE3", true,false,false,null);
34.         channel.queueDeclare("MESSAGE_QUEUE3", true,false,false,null);
35.         channel.queueDeclare("STATION_QUEUE3", true,false,false,null);
36.         //6、队列绑定交换机
37.         /**
38.          * 参数一：队列名
39.          * 参数二：交换机名字
40.          * 参数三：路由关键字，发布订阅模式写“”即可
41.          */
42.         channel.queueBind("MAIL_QUEUE3","exchange_topic","#.mail.#");
43.         channel.queueBind("MESSAGE_QUEUE3","exchange_topic","#.message.#");
44.         channel.queueBind("STATION_QUEUE3","exchange_topic","#.station.#");
45.         //7、发送消息
47.             channel.basicPublish("exchange_topic","mail.message.station",null,
48.                     ("你好，通配符模式，这是重要消息").getBytes(StandardCharsets.UTF_8));
50.             channel.basicPublish("exchange_topic","station",null,
51.                     ("你好，通配符模式，这是不太重要的消息").getBytes(StandardCharsets.UTF_8));
54.     }
55. }

复制代码

通配符模式消耗者

1. package com.tmh.mq.topic;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. public class ConsumerMail {
9.     public static void main(String[] args) throws IOException, TimeoutException {
10.         //1、创建连接工厂
11.         final ConnectionFactory connectionFactory = new ConnectionFactory();
12.         connectionFactory.setHost("192.168.66.8");
13.         connectionFactory.setPort(5672);
14.         connectionFactory.setUsername("tangminghao");
15.         connectionFactory.setPassword("tangminghao");
16.         connectionFactory.setVirtualHost("/");
17.         //2、建立连接
18.         final Connection connection = connectionFactory.newConnection();
19.         //3、建立信道
20.         final Channel channel = connection.createChannel();
21.         //4、监听队列
22.         channel.basicConsume("MAIL_QUEUE3",true,new DefaultConsumer(channel){
23.             @Override
24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
25.                 final String s = new String(body, "UTF-8");
26.                 System.out.println("邮件消息："+s);
27.             }
28.         });
30.     }
31. }
32. -----------------------------------------------------------------------------------------
33. public class ConsumerMessage {
34.     public static void main(String[] args) throws IOException, TimeoutException {
35.         //1、创建连接工厂
36.         final ConnectionFactory connectionFactory = new ConnectionFactory();
37.         connectionFactory.setHost("192.168.66.8");
38.         connectionFactory.setPort(5672);
39.         connectionFactory.setUsername("tangminghao");
40.         connectionFactory.setPassword("tangminghao");
41.         connectionFactory.setVirtualHost("/");
42.         //2、建立连接
43.         final Connection connection = connectionFactory.newConnection();
44.         //3、建立信道
45.         final Channel channel = connection.createChannel();
46.         //4、监听队列
47.         channel.basicConsume("MESSAGE_QUEUE3",true,new DefaultConsumer(channel){
48.             @Override
49.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
50.                 final String s = new String(body, "UTF-8");
51.                 System.out.println("短信消息："+s);
52.             }
53.         });
55.     }
56. }
57. -----------------------------------------------------------------------------------------
58. public class ConsumerStation {
59.     public static void main(String[] args) throws IOException, TimeoutException {
60.         //1、创建连接工厂
61.         final ConnectionFactory connectionFactory = new ConnectionFactory();
62.         connectionFactory.setHost("192.168.66.8");
63.         connectionFactory.setPort(5672);
64.         connectionFactory.setUsername("tangminghao");
65.         connectionFactory.setPassword("tangminghao");
66.         connectionFactory.setVirtualHost("/");
67.         //2、建立连接
68.         final Connection connection = connectionFactory.newConnection();
69.         //3、建立信道
70.         final Channel channel = connection.createChannel();
71.         //4、监听队列
72.         channel.basicConsume("STATION_QUEUE3",true,new DefaultConsumer(channel){
73.             @Override
74.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
75.                 final String s = new String(body, "UTF-8");
76.                 System.out.println("站内消息："+s);
77.             }
78.         });
81.     }
82. }

复制代码

在Java实现中有很多注释没有改过来，由于很多模式之间变化不大，所以就复制粘贴了

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4