diff --git a/src/main/java/cn/hezhaohui/mq/Recv.java b/src/main/java/cn/hezhaohui/mq/Recv.java new file mode 100644 index 0000000..f7f0920 --- /dev/null +++ b/src/main/java/cn/hezhaohui/mq/Recv.java @@ -0,0 +1,42 @@ +package cn.hezhaohui.mq; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; +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"; + + public static void main(String[] args) throws IOException, TimeoutException { + ConnectionFactory connectionFactory = new ConnectionFactory(); + Connection connection = connectionFactory.newConnection(); + Channel channel = connection.createChannel(); + + System.out.println(" [*] Waiting for message"); + + DeliverCallback deliverCallback = (consumerTag, delivery) -> { + String message = new String(delivery.getBody(), StandardCharsets.UTF_8); + System.out.println(" [x] Received '" + message + "'"); + try { + dowork(message); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } finally { + System.out.println(" [x] Done"); + } + }; + boolean autoAck = true; + channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {}); + } + + private static void dowork(String task) throws InterruptedException { + System.out.println(" [Working] " + task); + Thread.sleep(1000); + } +}