当前位置：首页 > news >正文

rabbitmq服务端消费端实例(direct和fanout模式)

news 2025/10/21 8:24:37

配置文件

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/***/
public class RabbitMQConnectionUtilLocal {public static Connection getConnection() throws Exception {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置RabbitMQ主机地址factory.setHost("localhost");// 设置RabbitMQ端口factory.setPort(5672);// 设置虚拟主机、用户名和密码factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true); // 自动重连// 返回连接return factory.newConnection();}
}

direct生产端


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;public class RoutingProducer {private final static String EXCHANGE_NAME = "direct_exchange";public static void main(String[] argv) throws Exception {// 获取到连接Connection connection = RabbitMQConnectionUtil.getConnection();// 从连接中创建通道Channel channel = connection.createChannel();// 声明交换器channel.exchangeDeclare(EXCHANGE_NAME, "direct");// 发送消息String message1 = "{\"flag\":0,\"idList\":[\"1\"]}";String message2 = "Info log message";String message3 = "Warning log message";channel.basicPublish(EXCHANGE_NAME, "error", null, message1.getBytes());System.out.println(" [x] Sent '" + message1 + "'");channel.basicPublish(EXCHANGE_NAME, "info", null, message2.getBytes());System.out.println(" [x] Sent '" + message2 + "'");channel.basicPublish(EXCHANGE_NAME, "warning", null, message3.getBytes());System.out.println(" [x] Sent '" + message3 + "'");// 关闭通道和连接channel.close();connection.close();}
}

fanout生产端


import com.rabbitmq.client.*;
import jzy.util.RabbitMQConnectionUtil;public class FanoutExchangeExample {private static final String EXCHANGE_NAME = "20250917test1";public static void main(String[] argv) throws Exception {Connection connection = RabbitMQConnectionUtilLocal.getConnection();Channel channel = connection.createChannel();// 声明Fanout类型的交换器，第三个参数是持久化标记channel.exchangeDeclare(EXCHANGE_NAME, "fanout",true);// 创建队列并绑定到交换器String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "");// 发送消息String message = "This is a log message";channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");channel.close();connection.close();}
}

消费端


package jzy.util;import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import com.rabbitmq.client.*;
import org.springframework.web.bind.annotation.RequestBody;import java.util.List;public class RabbitMQConsumer {
//    private final static String EXCHANGE_NAME = "direct_exchange";
//    private final static String QUEUE_NAME = "error_log_queue";private final static String EXCHANGE_NAME = "position_exchange";private final static String QUEUE_NAME = "LZF0101_queue";public static void main(String[] argv) throws Exception {// 获取到连接Connection connection = RabbitMQConnectionUtil.getConnection();// 从连接中创建通道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 绑定队列到交换器，并指定路由键channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "LZF0101_topic");// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws java.io.IOException {String message = new String(body, "UTF-8");JSONObject jsonObject = JsonUtil.getJsonToBean(replaceNbspWithSpace(message), JSONObject.class);try {switch (jsonObject.getStr("flag")) {default:break;}} catch (Exception e) {throw new RuntimeException(e);}System.out.println(" [x] Received  Log: '" + message + "'");}};// 监听队列channel.basicConsume(QUEUE_NAME, true, consumer);}/*** 正则过滤生产端信息特殊符号,过滤 NBSP*/public static String replaceNbspWithSpace(String str) {if (str == null) {return null;}return str.replaceAll("\\u00A0", " ");}
}