From 75fb4aa63f7962b4e4ea5c7af4d39ef84c99ebac Mon Sep 17 00:00:00 2001 From: Wonder Date: Fri, 14 Nov 2025 22:36:04 +0800 Subject: [PATCH] =?UTF-8?q?=20=F0=9F=94=84Update:=20=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E4=BF=A1=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/main/java/cn/hezhaohui/mq/Consume.java | 30 ++++++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 src/main/java/cn/hezhaohui/mq/Consume.java diff --git a/src/main/java/cn/hezhaohui/mq/Consume.java b/src/main/java/cn/hezhaohui/mq/Consume.java new file mode 100644 index 0000000..5936da3 --- /dev/null +++ b/src/main/java/cn/hezhaohui/mq/Consume.java @@ -0,0 +1,30 @@ +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.concurrent.TimeoutException; + +public class Consume { + private static final String EXCHANGE_NAME = "logs"; + + public static void main(String[] args) throws IOException, TimeoutException { + ConnectionFactory factory = new ConnectionFactory(); + Connection connection = factory.newConnection(); + Channel channel = connection.createChannel(); + + channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); + String queueName = channel.queueDeclare().getQueue(); + channel.queueBind(queueName, EXCHANGE_NAME, ""); + + DeliverCallback deliverCallback = (consumerTag, delivery) -> { + String message = new String(delivery.getBody(), StandardCharsets.UTF_8); + System.out.println("[Receive] " + message); + }; + channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {}); + } +}