diff --git a/src/main/java/cn/hezhaohui/MqApplicaion.java b/src/main/java/cn/hezhaohui/MqApplicaion.java index 577ef30..0605266 100644 --- a/src/main/java/cn/hezhaohui/MqApplicaion.java +++ b/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(); } } diff --git a/src/main/java/cn/hezhaohui/mq/Recv.java b/src/main/java/cn/hezhaohui/mq/Recv.java index f7f0920..48e6d58 100644 --- a/src/main/java/cn/hezhaohui/mq/Recv.java +++ b/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); } } diff --git a/src/main/java/cn/hezhaohui/mq/Send.java b/src/main/java/cn/hezhaohui/mq/Send.java index cfde3b4..ed54936 100644 --- a/src/main/java/cn/hezhaohui/mq/Send.java +++ b/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)) {