RabbitMQ

RabbitMQ























Messaging that just works — RabbitMQ
案例

pom.xml

1. <dependencies>
2.         <dependency>
3.             <groupId>com.rabbitmq</groupId>
4.             <artifactId>amqp-client</artifactId>
5.             <version>5.6.0</version>
6.         </dependency>
7.     </dependencies>
9.     <build>
10.         <plugins>
11.             <plugin>
12.                 <groupId>org.apache.maven.plugins</groupId>
13.                 <artifactId>maven-compiler-plugin</artifactId>
14.                 <version>3.8.0</version>
15.                 <configuration>
16.                     <source>1.8</source>
17.                     <target>1.8</target>
18.                 </configuration>
19.             </plugin>
20.         </plugins>
21.     </build>

复制代码

生产者

1. package com.www.mq;
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. * 生产者：发送消息
12. *
13. * @author Www
14. * @version 1.8
15. * @since 2023/2/15  20:41 星期三
16. */
17. public class ProducerHelloWorld {
18.    
19.     public static void main(String[] args) throws IOException, TimeoutException {
20.         // 1、创建连接工厂
21.         ConnectionFactory connectionFactory = new ConnectionFactory();
22.         // 2、设置参数
23.         // IP 默认值 localhost
24.         connectionFactory.setHost("192.168.36.100");
25.         // 端口 默认 5672
26.         connectionFactory.setPort(5672);
27.         // 虚拟机 默认值 /
28.         connectionFactory.setVirtualHost("/ljt");
29.         // 用户名 默认值 guest
30.         connectionFactory.setUsername("ljt");
31.         // 密码 默认值 guest
32.         connectionFactory.setPassword("ljt");
33.         
34.         
35.         // 3、创建连接 Connection : 受检异常——> 抛出
36.         Connection connection = connectionFactory.newConnection();
37.         
38.         // 4、创建通道 Channel
39.         Channel channel = connection.createChannel();
40.         
41.         // 5、创建队列 Queue
42.         /*
43.          queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
44.          参数：
45.              1. queue：队列名称
46.              2. durable:是否持久化，当mq重启之后，还在
47.              3. exclusive：
48.                  * 是否独占。只能有一个消费者监听这队列
49.                  * 当Connection关闭时，是否删除队列
50.                  *
51.              4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
52.              5. arguments：参数。
53.          */
54.         // 如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
55.         channel.queueDeclare(
56.                 // 队列名称
57.                 "hello_world",
58.                 // 是否持久化，当mq重启之后，还在
59.                 true,
60.                 // 是否独占。只能有一个消费者监听这队列 当Connection关闭时，是否删除队列
61.                 false,
62.                 // 是否自动删除。当没有Consumer时，自动删除掉
63.                 false,
64.                 // 参数
65.                 null
66.         );
67.         
68.          /*
69.              basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
70.             参数：
71.                 1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
72.                 2. routingKey：路由名称
73.                 3. props：配置信息
74.                 4. body：发送消息数据
75.          */
76.         // 发送消息内容
77.         String body = "HelloWorld~~~~~~~~~~";
78.         
79.         // 6、发送消息
80.         channel.basicPublish(
81.                 // 交换机名称。简单模式下交换机会使用默认的 ""
82.                 "",
83.                 // 路由名称
84.                 "hello_world",
85.                 // 配置信息
86.                 null,
87.                 // 发送消息数据
88.                 body.getBytes()
89.         );
90.         
91.         // 释放资源
92.         channel.close();
93.         connection.close();
94.         
95.     }
96. }

复制代码

消费者

1. package com.www.mq;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. /**
9. * 消费者：获取消息
10. *
11. * @author Www
12. * @version 1.8
13. * @since 2023/2/15  20:41 星期三
14. */
15. public class ConsumerHelloWorld {
16.    
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         // 1、创建连接工厂
19.         ConnectionFactory connectionFactory = new ConnectionFactory();
20.         // 2、设置参数
21.         // IP 默认值 localhost
22.         connectionFactory.setHost("192.168.36.100");
23.         // 端口 默认 5672
24.         connectionFactory.setPort(5672);
25.         // 虚拟机 默认值 /
26.         connectionFactory.setVirtualHost("/ljt");
27.         // 用户名 默认值 guest
28.         connectionFactory.setUsername("ljt");
29.         // 密码 默认值 guest
30.         connectionFactory.setPassword("ljt");
31.         
32.         
33.         // 3、创建连接 Connection : 受检异常——> 抛出
34.         Connection connection = connectionFactory.newConnection();
35.         
36.         // 4、创建通道 Channel
37.         Channel channel = connection.createChannel();
38.         
39.         // 5、创建队列 Queue
40.         /*
41.          queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
42.          参数：
43.              1. queue：队列名称
44.              2. durable:是否持久化，当mq重启之后，还在
45.              3. exclusive：
46.                  * 是否独占。只能有一个消费者监听这队列
47.                  * 当Connection关闭时，是否删除队列
48.                  *
49.              4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
50.              5. arguments：参数。
51.          */
52.         // 如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
53.         channel.queueDeclare(
54.                 // 队列名称
55.                 "hello_world",
56.                 // 是否持久化，当mq重启之后，还在
57.                 true,
58.                 // 是否独占。只能有一个消费者监听这队列 当Connection关闭时，是否删除队列
59.                 false,
60.                 // 是否自动删除。当没有Consumer时，自动删除掉
61.                 false,
62.                 // 参数
63.                 null);
64.         
65.          /*
66.             basicConsume(String queue, boolean autoAck, Consumer callback)
67.             参数：
68.                 1. queue：队列名称
69.                 2. autoAck：是否自动确认
70.                 3. callback：回调对象
71.          */
72.         // 6、接收消息
73.         Consumer consumer = new DefaultConsumer(channel) {
74.             /**
75.              * <p>
76.              *   回调方法，当收到消息后，会自动执行该方法
77.              *      1. consumerTag：标识
78.              *      2. envelope：获取一些信息，交换机，路由key...
79.              *      3. properties:配置信息
80.              *      4. body：数据
81.              * </p>
82.              */
83.             @Override
84.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
85.                 System.out.println("consumerTag：" + consumerTag);
86.                 System.out.println("Exchange：" + envelope.getExchange());
87.                 System.out.println("RoutingKey：" + envelope.getRoutingKey());
88.                 System.out.println("properties：" + properties);
89.                 System.out.println("body：" + new String(body));
90.             }
91.         };
92.         channel.basicConsume(
93.                 // 队列名称
94.                 "hello_world",
95.                 // 是否自动确认
96.                 true,
97.                 // 回调对象
98.                 consumer
99.         );
100.         
101.         // 不需要关闭资源
102.     }
103. }

复制代码

工作队列


生产者

1. package com.www.mq;
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. * 生产者：发送消息
12. *
13. * @author Www
14. * @version 1.8
15. * @since 2023/2/15  20:41 星期三
16. */
17. public class ProducerWorkQueues {
18.    
19.     public static void main(String[] args) throws IOException, TimeoutException {
20.         // 1、创建连接工厂
21.         ConnectionFactory connectionFactory = new ConnectionFactory();
22.         // 2、设置参数
23.         // IP 默认值 localhost
24.         connectionFactory.setHost("192.168.36.100");
25.         // 端口 默认 5672
26.         connectionFactory.setPort(5672);
27.         // 虚拟机 默认值 /
28.         connectionFactory.setVirtualHost("/ljt");
29.         // 用户名 默认值 guest
30.         connectionFactory.setUsername("ljt");
31.         // 密码 默认值 guest
32.         connectionFactory.setPassword("ljt");
33.         
34.         
35.         // 3、创建连接 Connection : 受检异常——> 抛出
36.         Connection connection = connectionFactory.newConnection();
37.         
38.         // 4、创建通道 Channel
39.         Channel channel = connection.createChannel();
40.         
41.         // 5、创建队列 Queue
42.         /*
43.          queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
44.          参数：
45.              1. queue：队列名称
46.              2. durable:是否持久化，当mq重启之后，还在
47.              3. exclusive：
48.                  * 是否独占。只能有一个消费者监听这队列
49.                  * 当Connection关闭时，是否删除队列
50.                  *
51.              4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
52.              5. arguments：参数。
53.          */
54.         // 如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
55.         channel.queueDeclare(
56.                 // 队列名称
57.                 "WorkQueue",
58.                 // 是否持久化，当mq重启之后，还在
59.                 true,
60.                 // 是否独占。只能有一个消费者监听这队列 当Connection关闭时，是否删除队列
61.                 false,
62.                 // 是否自动删除。当没有Consumer时，自动删除掉
63.                 false,
64.                 // 参数
65.                 null
66.         );
67.         
68.          /*
69.              basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
70.             参数：
71.                 1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
72.                 2. routingKey：路由名称
73.                 3. props：配置信息
74.                 4. body：发送消息数据
75.          */
76.         for (int i = 0; i < 10; i++) {
77.             // 发送消息内容
78.             String body = "WorkQueue~~~~~~~~~~" + i;
79.             
80.             // 6、发送消息
81.             channel.basicPublish(
82.                     // 交换机名称。简单模式下交换机会使用默认的 ""
83.                     "",
84.                     // 路由名称
85.                     "WorkQueue",
86.                     // 配置信息
87.                     null,
88.                     // 发送消息数据
89.                     body.getBytes()
90.             );
91.         }
92.         
93.         
94.         // 释放资源
95.         channel.close();
96.         connection.close();
97.         
98.     }
99. }

复制代码

消费者: 两个

1. package com.www.mq;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. /**
9. * 消费者：获取消息
10. *
11. * @author Www
12. * @version 1.8
13. * @since 2023/2/15  20:41 星期三
14. */
15. public class ConsumerWorkQueues1 {
16.    
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         // 1、创建连接工厂
19.         ConnectionFactory connectionFactory = new ConnectionFactory();
20.         // 2、设置参数
21.         // IP 默认值 localhost
22.         connectionFactory.setHost("192.168.36.100");
23.         // 端口 默认 5672
24.         connectionFactory.setPort(5672);
25.         // 虚拟机 默认值 /
26.         connectionFactory.setVirtualHost("/ljt");
27.         // 用户名 默认值 guest
28.         connectionFactory.setUsername("ljt");
29.         // 密码 默认值 guest
30.         connectionFactory.setPassword("ljt");
31.         
32.         
33.         // 3、创建连接 Connection : 受检异常——> 抛出
34.         Connection connection = connectionFactory.newConnection();
35.         
36.         // 4、创建通道 Channel
37.         Channel channel = connection.createChannel();
38.         
39.         // 5、创建队列 Queue
40.         /*
41.          queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
42.          参数：
43.              1. queue：队列名称
44.              2. durable:是否持久化，当mq重启之后，还在
45.              3. exclusive：
46.                  * 是否独占。只能有一个消费者监听这队列
47.                  * 当Connection关闭时，是否删除队列
48.                  *
49.              4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉
50.              5. arguments：参数。
51.          */
52.         // 如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
53.         channel.queueDeclare(
54.                 // 队列名称
55.                 "WorkQueue",
56.                 // 是否持久化，当mq重启之后，还在
57.                 true,
58.                 // 是否独占。只能有一个消费者监听这队列 当Connection关闭时，是否删除队列
59.                 false,
60.                 // 是否自动删除。当没有Consumer时，自动删除掉
61.                 false,
62.                 // 参数
63.                 null);
64.         
65.          /*
66.             basicConsume(String queue, boolean autoAck, Consumer callback)
67.             参数：
68.                 1. queue：队列名称
69.                 2. autoAck：是否自动确认
70.                 3. callback：回调对象
71.          */
72.         // 6、接收消息
73.         Consumer consumer = new DefaultConsumer(channel) {
74.             /**
75.              * <p>
76.              *   回调方法，当收到消息后，会自动执行该方法
77.              *      1. consumerTag：标识
78.              *      2. envelope：获取一些信息，交换机，路由key...
79.              *      3. properties:配置信息
80.              *      4. body：数据
81.              * </p>
82.              */
83.             @Override
84.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
85.                 System.out.println("consumerTag：" + consumerTag);
86.                 System.out.println("Exchange：" + envelope.getExchange());
87.                 System.out.println("RoutingKey：" + envelope.getRoutingKey());
88.                 System.out.println("properties：" + properties);
89.                 System.out.println("body：" + new String(body));
90.             }
91.         };
92.         channel.basicConsume(
93.                 // 队列名称
94.                 "WorkQueue",
95.                 // 是否自动确认
96.                 true,
97.                 // 回调对象
98.                 consumer
99.         );
100.         
101.         // 不需要关闭资源
102.     }
103. }

复制代码

订阅模式


生产者

1. package com.www.mq;
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.util.concurrent.TimeoutException;
11. /**
12. * 生产者：发送消息
13. *
14. * @author Www
15. * @version 1.8
16. * @since 2023/2/15  20:41 星期三
17. */
18. public class ProducerPubSub {
19.    
20.     public static void main(String[] args) throws IOException, TimeoutException {
21.         // 1、创建连接工厂
22.         ConnectionFactory connectionFactory = new ConnectionFactory();
23.         // 2、设置参数
24.         // IP 默认值 localhost
25.         connectionFactory.setHost("192.168.36.100");
26.         // 端口 默认 5672
27.         connectionFactory.setPort(5672);
28.         // 虚拟机 默认值 /
29.         connectionFactory.setVirtualHost("/ljt");
30.         // 用户名 默认值 guest
31.         connectionFactory.setUsername("ljt");
32.         // 密码 默认值 guest
33.         connectionFactory.setPassword("ljt");
34.         
35.         
36.         // 3、创建连接 Connection : 受检异常——> 抛出
37.         Connection connection = connectionFactory.newConnection();
38.         
39.         // 4、创建通道 Channel
40.         Channel channel = connection.createChannel();
41.         
42.         // 5、创建交换机
43.         /*
44.            exchangeDeclare(
45.             String exchange,BuiltinExchangeType type,
46.             boolean durable, boolean autoDelete,
47.             boolean internal, Map<String, Object> arguments
48.             )
49.            参数：
50.             1. exchange:交换机名称
51.             2. type:交换机类型
52.                 DIRECT("direct"),：定向
53.                 FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
54.                 TOPIC("topic"),通配符的方式
55.                 HEADERS("headers");参数匹配
56.             3. durable:是否持久化
57.             4. autoDelete:自动删除
58.             5. internal：内部使用。 一般false
59.             6. arguments：参数
60.         */
61.         String exchangeName = "test_fanout";
62.         channel.exchangeDeclare(
63.                 // 交换机名称
64.                 exchangeName,
65.                 // type:交换机类型
66.                 //   DIRECT("direct"),：定向
67.                 //   FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
68.                 //   TOPIC("topic"),通配符的方式
69.                 //   HEADERS("headers");参数匹配
70.                 BuiltinExchangeType.FANOUT,
71.                 // 是否持久化
72.                 true,
73.                 // 内部使用
74.                 false,
75.                 // 参数
76.                 null
77.         
78.         );
79.         
80.         // 6、创建队列
81.         String queue1Name = "test_fanout_queue1";
82.         String queue2Name = "test_fanout_queue2";
83.         channel.queueDeclare(
84.                 // 队列名
85.                 queue1Name,
86.                 // 是否持久
87.                 true,
88.                 // 是否独占
89.                 false,
90.                 //是否自动删除
91.                 false,
92.                 // 参数
93.                 null
94.         );
95.         channel.queueDeclare(
96.                 // 队列名
97.                 queue2Name,
98.                 // 是否持久
99.                 true,
100.                 // 是否独占
101.                 false,
102.                 //是否自动删除
103.                 false,
104.                 // 参数
105.                 null
106.         );
107.         
108.         // 7、绑定队列和交换机
109.          /*
110.             queueBind(String queue, String exchange, String routingKey)
111.             参数：
112.                 1. queue：队列名称
113.                 2. exchange：交换机名称
114.                 3. routingKey：路由键，绑定规则
115.                     如果交换机的类型为fanout ，routingKey设置为""
116.          */
117.         channel.queueBind(
118.                 // 队列名
119.                 queue1Name,
120.                 // 交换机名
121.                 exchangeName,
122.                 //  routingKey：路由键，绑定规则
123.                 //    如果交换机的类型为fanout ，routingKey设置为""
124.                 ""
125.         );
126.         channel.queueBind(
127.                 // 队列名
128.                 queue2Name,
129.                 // 交换机名
130.                 exchangeName,
131.                 //  routingKey：路由键，绑定规则
132.                 //    如果交换机的类型为fanout ，routingKey设置为""
133.                 ""
134.         );
135.         
136.          /*
137.              basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
138.             参数：
139.                 1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
140.                 2. routingKey：路由名称
141.                 3. props：配置信息
142.                 4. body：发送消息数据
143.          */
144.         // 发送消息内容
145.         String body = "日志信息：张三调用了findAll方法...日志级别：info...";
146.         // 8、发送消息
147.         channel.basicPublish(
148.                 // 交换机名称。简单模式下交换机会使用默认的 ""
149.                 exchangeName,
150.                 // 路由名称 ：如果交换机的类型为fanout ，routingKey设置为""
151.                 "",
152.                 // 配置信息
153.                 null,
154.                 // 发送消息数据
155.                 body.getBytes()
156.         );
157.         
158.         
159.         // 9、释放资源
160.         channel.close();
161.         connection.close();
162.         
163.     }
164. }

复制代码

消费者1

1. package com.www.mq;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. /**
9. * 消费者：获取消息
10. *
11. * @author Www
12. * @version 1.8
13. * @since 2023/2/15  20:41 星期三
14. */
15. public class ConsumerPubSub1 {
16.    
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         // 1、创建连接工厂
19.         ConnectionFactory connectionFactory = new ConnectionFactory();
20.         // 2、设置参数
21.         // IP 默认值 localhost
22.         connectionFactory.setHost("192.168.36.100");
23.         // 端口 默认 5672
24.         connectionFactory.setPort(5672);
25.         // 虚拟机 默认值 /
26.         connectionFactory.setVirtualHost("/ljt");
27.         // 用户名 默认值 guest
28.         connectionFactory.setUsername("ljt");
29.         // 密码 默认值 guest
30.         connectionFactory.setPassword("ljt");
31.         
32.         
33.         // 3、创建连接 Connection : 受检异常——> 抛出
34.         Connection connection = connectionFactory.newConnection();
35.         
36.         // 4、创建通道 Channel
37.         Channel channel = connection.createChannel();
38.         
39.         // 队列名
40.         String queue1Name = "test_fanout_queue1";
41.       
42.         
43.          /*
44.             basicConsume(String queue, boolean autoAck, Consumer callback)
45.             参数：
46.                 1. queue：队列名称
47.                 2. autoAck：是否自动确认
48.                 3. callback：回调对象
49.          */
50.         // 5、接收消息
51.         Consumer consumer = new DefaultConsumer(channel) {
52.             /**
53.              * <p>
54.              *   回调方法，当收到消息后，会自动执行该方法
55.              *      1. consumerTag：标识
56.              *      2. envelope：获取一些信息，交换机，路由key...
57.              *      3. properties:配置信息
58.              *      4. body：数据
59.              * </p>
60.              */
61.             @Override
62.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
63.                 System.out.println("consumerTag：" + consumerTag);
64.                 System.out.println("Exchange：" + envelope.getExchange());
65.                 System.out.println("RoutingKey：" + envelope.getRoutingKey());
66.                 System.out.println("properties：" + properties);
67.                 System.out.println("body：" + new String(body));
68.             }
69.         };
70.         channel.basicConsume(
71.                 // 队列名称
72.                 queue1Name,
73.                 // 是否自动确认
74.                 true,
75.                 // 回调对象
76.                 consumer
77.         );
78.         
79.         // 不需要关闭资源
80.     }
81. }

复制代码

消费者2

1. package com.www.mq;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. /**
9. * 消费者：获取消息
10. *
11. * @author Www
12. * @version 1.8
13. * @since 2023/2/15  20:41 星期三
14. */
15. public class ConsumerPubSub2 {
16.    
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         // 1、创建连接工厂
19.         ConnectionFactory connectionFactory = new ConnectionFactory();
20.         // 2、设置参数
21.         // IP 默认值 localhost
22.         connectionFactory.setHost("192.168.36.100");
23.         // 端口 默认 5672
24.         connectionFactory.setPort(5672);
25.         // 虚拟机 默认值 /
26.         connectionFactory.setVirtualHost("/ljt");
27.         // 用户名 默认值 guest
28.         connectionFactory.setUsername("ljt");
29.         // 密码 默认值 guest
30.         connectionFactory.setPassword("ljt");
31.         
32.         
33.         // 3、创建连接 Connection : 受检异常——> 抛出
34.         Connection connection = connectionFactory.newConnection();
35.         
36.         // 4、创建通道 Channel
37.         Channel channel = connection.createChannel();
38.         
39.         // 队列名
40.         String queue2Name = "test_fanout_queue2";
41.       
42.         
43.          /*
44.             basicConsume(String queue, boolean autoAck, Consumer callback)
45.             参数：
46.                 1. queue：队列名称
47.                 2. autoAck：是否自动确认
48.                 3. callback：回调对象
49.          */
50.         // 5、接收消息
51.         Consumer consumer = new DefaultConsumer(channel) {
52.             /**
53.              * <p>
54.              *   回调方法，当收到消息后，会自动执行该方法
55.              *      1. consumerTag：标识
56.              *      2. envelope：获取一些信息，交换机，路由key...
57.              *      3. properties:配置信息
58.              *      4. body：数据
59.              * </p>
60.              */
61.             @Override
62.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
63.                 System.out.println("consumerTag：" + consumerTag);
64.                 System.out.println("Exchange：" + envelope.getExchange());
65.                 System.out.println("RoutingKey：" + envelope.getRoutingKey());
66.                 System.out.println("properties：" + properties);
67.                 System.out.println("body：" + new String(body));
68.             }
69.         };
70.         channel.basicConsume(
71.                 // 队列名称
72.                 queue2Name,
73.                 // 是否自动确认
74.                 true,
75.                 // 回调对象
76.                 consumer
77.         );
78.         
79.         // 不需要关闭资源
80.     }
81. }

复制代码

Routing 路由模式



生产者

1. package com.www.mq;
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.util.concurrent.TimeoutException;
11. /**
12. * Routing 工作模式
13. * <p>
14. * 生产者：发送消息
15. *
16. * @author Www
17. * @version 1.8
18. * @since 2023/2/15  20:41 星期三
19. */
20. public class ProducerRouting {
21.    
22.     public static void main(String[] args) throws IOException, TimeoutException {
23.         // 1、创建连接工厂
24.         ConnectionFactory connectionFactory = new ConnectionFactory();
25.         // 2、设置参数
26.         // IP 默认值 localhost
27.         connectionFactory.setHost("192.168.36.100");
28.         // 端口 默认 5672
29.         connectionFactory.setPort(5672);
30.         // 虚拟机 默认值 /
31.         connectionFactory.setVirtualHost("/ljt");
32.         // 用户名 默认值 guest
33.         connectionFactory.setUsername("ljt");
34.         // 密码 默认值 guest
35.         connectionFactory.setPassword("ljt");
36.         
37.         
38.         // 3、创建连接 Connection : 受检异常——> 抛出
39.         Connection connection = connectionFactory.newConnection();
40.         
41.         // 4、创建通道 Channel
42.         Channel channel = connection.createChannel();
43.         
44.         // 5、创建交换机
45.         /*
46.            exchangeDeclare(
47.             String exchange,BuiltinExchangeType type,
48.             boolean durable, boolean autoDelete,
49.             boolean internal, Map<String, Object> arguments
50.             )
51.            参数：
52.             1. exchange:交换机名称
53.             2. type:交换机类型
54.                 DIRECT("direct"),：定向
55.                 FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
56.                 TOPIC("topic"),通配符的方式
57.                 HEADERS("headers");参数匹配
58.             3. durable:是否持久化
59.             4. autoDelete:自动删除
60.             5. internal：内部使用。 一般false
61.             6. arguments：参数
62.         */
63.         String exchangeName = "test_direct";
64.         channel.exchangeDeclare(
65.                 // 交换机名称
66.                 exchangeName,
67.                 // type:交换机类型
68.                 //   DIRECT("direct"),：定向
69.                 //   FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
70.                 //   TOPIC("topic"),通配符的方式
71.                 //   HEADERS("headers");参数匹配
72.                 BuiltinExchangeType.DIRECT,
73.                 // 是否持久化
74.                 true,
75.                 // 内部使用
76.                 false,
77.                 // 参数
78.                 null
79.         
80.         );
81.         
82.         // 6、创建队列
83.         String queue1Name = "test_direct_queue1";
84.         String queue2Name = "test_direct_queue2";
85.         channel.queueDeclare(
86.                 // 队列名
87.                 queue1Name,
88.                 // 是否持久
89.                 true,
90.                 // 是否独占
91.                 false,
92.                 //是否自动删除
93.                 false,
94.                 // 参数
95.                 null
96.         );
97.         channel.queueDeclare(
98.                 // 队列名
99.                 queue2Name,
100.                 // 是否持久
101.                 true,
102.                 // 是否独占
103.                 false,
104.                 //是否自动删除
105.                 false,
106.                 // 参数
107.                 null
108.         );
109.         
110.         // 7、绑定队列和交换机
111.          /*
112.             queueBind(String queue, String exchange, String routingKey)
113.             参数：
114.                 1. queue：队列名称
115.                 2. exchange：交换机名称
116.                 3. routingKey：路由键，绑定规则
117.                     如果交换机的类型为fanout ，routingKey设置为""
118.          */
119.         // 队列1
120.         channel.queueBind(
121.                 // 队列名
122.                 queue1Name,
123.                 // 交换机名
124.                 exchangeName,
125.                 //  routingKey：路由键，绑定规则
126.                 //    如果交换机的类型为fanout ，routingKey设置为""
127.                 "error"
128.         );
129.         // 队列2
130.         channel.queueBind(
131.                 // 队列名
132.                 queue2Name,
133.                 // 交换机名
134.                 exchangeName,
135.                 //  routingKey：路由键，绑定规则
136.                 //    如果交换机的类型为fanout ，routingKey设置为""
137.                 "info"
138.         );
139.         channel.queueBind(
140.                 // 队列名
141.                 queue2Name,
142.                 // 交换机名
143.                 exchangeName,
144.                 //  routingKey：路由键，绑定规则
145.                 //    如果交换机的类型为fanout ，routingKey设置为""
146.                 "error"
147.         );
148.         channel.queueBind(
149.                 // 队列名
150.                 queue2Name,
151.                 // 交换机名
152.                 exchangeName,
153.                 //  routingKey：路由键，绑定规则
154.                 //    如果交换机的类型为fanout ，routingKey设置为""
155.                 "waring"
156.         );
157.         
158.          /*
159.              basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
160.             参数：
161.                 1. exchange：交换机名称。简单模式下交换机会使用默认的 ""
162.                 2. routingKey：路由名称
163.                 3. props：配置信息
164.                 4. body：发送消息数据
165.          */
166.         // 发送消息内容
167.         String body = "日志信息：张三调用了delete方法...警告。。。日志级别：waring...";
168.         // 8、发送消息
169.         channel.basicPublish(
170.                 // 交换机名称。简单模式下交换机会使用默认的 ""
171.                 exchangeName,
172.                 // 路由名称 ：如果交换机的类型为fanout ，routingKey设置为""
173.                 "waring",
174.                 // 配置信息
175.                 null,
176.                 // 发送消息数据
177.                 body.getBytes()
178.         );
179.         
180.         
181.         // 9、释放资源
182.         channel.close();
183.         connection.close();
184.         
185.     }
186. }

复制代码

消费者 1

1. package com.www.mq;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. /**
9. * 消费者：获取消息
10. *
11. * @author Www
12. * @version 1.8
13. * @since 2023/2/15  20:41 星期三
14. */
15. public class ConsumerRouting1 {
16.    
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         // 1、创建连接工厂
19.         ConnectionFactory connectionFactory = new ConnectionFactory();
20.         // 2、设置参数
21.         // IP 默认值 localhost
22.         connectionFactory.setHost("192.168.36.100");
23.         // 端口 默认 5672
24.         connectionFactory.setPort(5672);
25.         // 虚拟机 默认值 /
26.         connectionFactory.setVirtualHost("/ljt");
27.         // 用户名 默认值 guest
28.         connectionFactory.setUsername("ljt");
29.         // 密码 默认值 guest
30.         connectionFactory.setPassword("ljt");
31.         
32.         
33.         // 3、创建连接 Connection : 受检异常——> 抛出
34.         Connection connection = connectionFactory.newConnection();
35.         
36.         // 4、创建通道 Channel
37.         Channel channel = connection.createChannel();
38.         
39.         // 队列名
40.         String queue2Name = "test_direct_queue1";
41.          /*
42.             basicConsume(String queue, boolean autoAck, Consumer callback)
43.             参数：
44.                 1. queue：队列名称
45.                 2. autoAck：是否自动确认
46.                 3. callback：回调对象
47.          */
48.         // 5、接收消息
49.         Consumer consumer = new DefaultConsumer(channel) {
50.             /**
51.              * <p>
52.              *   回调方法，当收到消息后，会自动执行该方法
53.              *      1. consumerTag：标识
54.              *      2. envelope：获取一些信息，交换机，路由key...
55.              *      3. properties:配置信息
56.              *      4. body：数据
57.              * </p>
58.              */
59.             @Override
60.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
61.                 System.out.println("consumerTag：" + consumerTag);
62.                 System.out.println("Exchange：" + envelope.getExchange());
63.                 System.out.println("RoutingKey：" + envelope.getRoutingKey());
64.                 System.out.println("properties：" + properties);
65.                 System.out.println("body：" + new String(body));
66.             }
67.         };
68.         channel.basicConsume(
69.                 // 队列名称
70.                 queue2Name,
71.                 // 是否自动确认
72.                 true,
73.                 // 回调对象
74.                 consumer
75.         );
76.         
77.         // 不需要关闭资源
78.     }
79. }

复制代码

消费者2

1. package com.www.mq;
3. import com.rabbitmq.client.*;
5. import java.io.IOException;
6. import java.util.concurrent.TimeoutException;
8. /**
9. * 消费者：获取消息
10. *
11. * @author Www
12. * @version 1.8
13. * @since 2023/2/15  20:41 星期三
14. */
15. public class ConsumerRouting2 {
16.    
17.     public static void main(String[] args) throws IOException, TimeoutException {
18.         // 1、创建连接工厂
19.         ConnectionFactory connectionFactory = new ConnectionFactory();
20.         // 2、设置参数
21.         // IP 默认值 localhost
22.         connectionFactory.setHost("192.168.36.100");
23.         // 端口 默认 5672
24.         connectionFactory.setPort(5672);
25.         // 虚拟机 默认值 /
26.         connectionFactory.setVirtualHost("/ljt");
27.         // 用户名 默认值 guest
28.         connectionFactory.setUsername("ljt");
29.         // 密码 默认值 guest
30.         connectionFactory.setPassword("ljt");
31.         
32.         
33.         // 3、创建连接 Connection : 受检异常——> 抛出
34.         Connection connection = connectionFactory.newConnection();
35.         
36.         // 4、创建通道 Channel
37.         Channel channel = connection.createChannel();
38.         
39.         // 队列名
40.         String queue2Name = "test_direct_queue2";
41.          /*
42.             basicConsume(String queue, boolean autoAck, Consumer callback)
43.             参数：
44.                 1. queue：队列名称
45.                 2. autoAck：是否自动确认
46.                 3. callback：回调对象
47.          */
48.         // 5、接收消息
49.         Consumer consumer = new DefaultConsumer(channel) {
50.             /**
51.              * <p>
52.              *   回调方法，当收到消息后，会自动执行该方法
53.              *      1. consumerTag：标识
54.              *      2. envelope：获取一些信息，交换机，路由key...
55.              *      3. properties:配置信息
56.              *      4. body：数据
57.              * </p>
58.              */
59.             @Override
60.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
61.                 System.out.println("consumerTag：" + consumerTag);
62.                 System.out.println("Exchange：" + envelope.getExchange());
63.                 System.out.println("RoutingKey：" + envelope.getRoutingKey());
64.                 System.out.println("properties：" + properties);
65.                 System.out.println("body：" + new String(body));
66.             }
67.         };
68.         channel.basicConsume(
69.                 // 队列名称
70.                 queue2Name,
71.                 // 是否自动确认
72.                 true,
73.                 // 回调对象
74.                 consumer
75.         );
76.         
77.         // 不需要关闭资源
78.     }
79. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！