🔄Update: 队列、消息持久化 & 公平分发

src/main/java/cn/hezhaohui/MqApplicaion.java

@@ -1,11 +1,30 @@
 package cn.hezhaohui;
 
+import cn.hezhaohui.mq.Recv;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
-@SpringBootApplication
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
 public class MqApplicaion {
-    public static void main(String[] args) {
-        SpringApplication.run(MqApplicaion.class, args);
+    public static void main(String[] args) throws IOException, TimeoutException {
+        new Thread(() -> {
+            try {
+                Recv.work("1");
+            } catch (IOException | TimeoutException e) {
+                throw new RuntimeException(e);
+            }
+
+        }).start();
+
+        new Thread(() -> {
+            try {
+                Recv.work("2");
+            } catch (IOException | TimeoutException e) {
+                throw new RuntimeException(e);
+            }
+
+        }).start();
     }
 }

src/main/java/cn/hezhaohui/mq/Recv.java

@@ -7,17 +7,16 @@ import com.rabbitmq.client.DeliverCallback;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.Random;
 import java.util.concurrent.TimeoutException;
 
 public class Recv {
-    private final static String QUEUE_NAME = "hello";
+    private final static String QUEUE_NAME = "task";
 
-    public static void main(String[] args) throws IOException, TimeoutException {
+    public static void work(String id) throws IOException, TimeoutException {
         ConnectionFactory connectionFactory = new ConnectionFactory();
         Connection connection = connectionFactory.newConnection();
         Channel channel = connection.createChannel();
-
+        channel.basicQos(1);
         System.out.println(" [*] Waiting for message");
 
         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
@@ -29,6 +28,7 @@ public class Recv {
                 throw new RuntimeException(e);
             } finally {
                 System.out.println(" [x] Done");
+                System.out.println("== FROM " + id + " ==");
             }
         };
         boolean autoAck = true;
@@ -37,6 +37,6 @@ public class Recv {
 
     private static void dowork(String task) throws InterruptedException {
         System.out.println(" [Working] " + task);
-        Thread.sleep(1000);
+        Thread.sleep(5000);
     }
 }

src/main/java/cn/hezhaohui/mq/Send.java

@@ -3,26 +3,30 @@ package cn.hezhaohui.mq;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.MessageProperties;
 
 import java.io.IOException;
 import java.util.Scanner;
 import java.util.concurrent.TimeoutException;
 
 public class Send {
-    private final static String QUEUE_NAME = "hello";
+    private final static String QUEUE_NAME = "task";
 
     public static void main(String[] args) {
         ConnectionFactory connectionFactory = new ConnectionFactory();
         try (Connection connection = connectionFactory.newConnection();
              Channel channel = connection.createChannel();
              Scanner scanner = new Scanner(System.in)){
-            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+            boolean durable = true;
+            channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
             System.out.println(" === Now you can send message ===");
 
             while (true) {
                 System.out.println("[Send]:");
                 String message = scanner.nextLine();
-                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
+                channel.basicPublish("", QUEUE_NAME,
+                        MessageProperties.PERSISTENT_TEXT_PLAIN,
+                        message.getBytes());
                 System.out.println(" [x] Sent '" + message + "'");
 
                 if ("exit".equalsIgnoreCase(message)) {