初始化 rabbitmq 示例

Author: YunaiV

lab-04/lab-04-rabbitmq-native/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/RabbitMQConsumer.java

@@ -8,41 +8,28 @@
 
 public class RabbitMQConsumer {
 
-    private static final String EXCHANGE_NAME = "exchange_demo";
-    private static final String ROUTING_KEY = "routingkey_demo";
-    private static final String QUEUE_NAME = "queue_demo";
-    private static final String IP_ADDRESS = "127.0.0.1";
-    public static final Integer PORT = 5672;
-
     public static void main(String[] args) throws IOException, TimeoutException {
         // 创建连接
-        ConnectionFactory factory = new ConnectionFactory();
-        factory.setHost(IP_ADDRESS);
-        factory.setPort(PORT);
-        factory.setUsername("guest");
-        factory.setPassword("guest");
-        Connection connection = factory.newConnection();
+        Connection connection = RabbitMQProducer.getConnection();
+
         // 创建信道
         final Channel channel = connection.createChannel();
         channel.basicQos(64); // 设置客户端最多接收未被 ack 的消息数量为 64 。
-        // 创建交换器：direct、持久化、不自动删除
-        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
 
         // 创建消费者
         Consumer consumer = new DefaultConsumer(channel) {
 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                // 打印日志
                 System.out.println("线程：" + Thread.currentThread() +  "：消息：" + new String(body));
-                try {
-                    TimeUnit.SECONDS.sleep(1);
-                } catch (InterruptedException ignore) {
-                }
+                // ack 消息已经消费
                 channel.basicAck(envelope.getDeliveryTag(), false);
             }
 
         };
-        channel.basicConsume(QUEUE_NAME, consumer);
+        // 订阅消费 QUEUE_NAME 队列
+        channel.basicConsume(RabbitMQProducer.QUEUE_NAME, consumer);
 
         // 关闭
         try {

lab-04/lab-04-rabbitmq-native/src/main/java/cn/iocoder/springboot/lab04/rabbitmqdemo/RabbitMQProducer.java

@@ -10,33 +10,27 @@
 
 public class RabbitMQProducer {
 
+    private static final String IP_ADDRESS = "127.0.0.1";
+    private static final Integer PORT = 5672;
+    private static final String USERNAME = "guest";
+    private static final String PASSWORD = "guest";
+
     private static final String EXCHANGE_NAME = "exchange_demo";
     private static final String ROUTING_KEY = "routingkey_demo";
-    private static final String QUEUE_NAME = "queue_demo";
-    private static final String IP_ADDRESS = "127.0.0.1";
-    public static final Integer PORT = 5672;
+    public static final String QUEUE_NAME = "queue_demo"; // 只有 QUEUE_NAME 需要共享给 RabbitMQConsumer
 
     public static void main(String[] args) throws IOException, TimeoutException {
         // 创建连接
-        ConnectionFactory factory = new ConnectionFactory();
-        factory.setHost(IP_ADDRESS);
-        factory.setPort(PORT);
-        factory.setUsername("guest");
-        factory.setPassword("guest");
-        Connection connection = factory.newConnection();
+        Connection connection = getConnection();
+
         // 创建信道
         Channel channel = connection.createChannel();
-        // 创建交换器：direct、持久化、不自动删除
-        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
-        // 创建队列：持久化、非排他、非自动删除的队列
-        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
-//        channel.queueDeclare(QUEUE_NAME + "_2", true, false, false, null);
-        // 将交换器与队列通过路由键绑定
-        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
-//        channel.queueBind(QUEUE_NAME + "_2", EXCHANGE_NAME, ROUTING_KEY);
 
-        // 发送一条消息
-        for (int i = 0; i < 1000; i++) {
+        // 初始化测试用的 Exchange 和 Queue
+        initExchangeAndQueue(channel);
+
+        // 发送 3 条消息
+        for (int i = 0; i < 3; i++) {
             String message = "Hello World" + i;
             channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
         }
@@ -46,4 +40,26 @@ public static void main(String[] args) throws IOException, TimeoutException {
         connection.close();
     }
 
+    public static Connection getConnection() throws IOException, TimeoutException {
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost(IP_ADDRESS);
+        factory.setPort(PORT);
+        factory.setUsername(USERNAME);
+        factory.setPassword(PASSWORD);
+        return factory.newConnection();
+    }
+
+    // 创建 RabbitMQ Exchange 和 Queue ，然后使用 ROUTING_KEY 路由键将两者绑定。
+    // 该步骤，其实可以在 RabbitMQ Management 上操作，并不一定需要在代码中
+    private static void initExchangeAndQueue(Channel channel) throws IOException {
+        // 创建交换器：direct、持久化、不自动删除
+        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
+
+        // 创建队列：持久化、非排他、非自动删除的队列
+        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
+
+        // 将交换器与队列通过路由键绑定
+        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
+    }
+
 }