RabbitMQ通过代码实现交换机分发模式

news/2025/3/30 16:16:13/文章来源:https://www.cnblogs.com/5ran2yl/p/18795042

首先导入rabbitMQ的依赖：

  <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.25.0</version></dependency>

一.fanout模式实现

图示：

在代码实现中，我们只需要关注生产者和消费者，因为交换机的创建和队列的绑定一般都是由图形化界面完成的，故而使用Java代码时候，就只需要实现生产者和消费者了

producer.Java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 6： 准备发送消息的内容String message = "你好，回忆也交给时间";String  exchangeName = "fanout_exchange";String routingKey = "";// 7: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称/routingkey// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish(exchangeName, routingKey, null, message.getBytes());System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

Consumer.java：

import com.rabbitmq.client.*;import java.io.IOException;public class Consumer {private static Runnable runnable = () -> {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");//获取队列的名称final String queueName = Thread.currentThread().getName();Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在，则会创建*  Rabbitmq不允许创建两个相同的队列名称，否则会报错。**  @params1： queue 队列的名称*  @params2： durable 队列是否持久化*  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭*  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。*  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。* */// 这里如果queue已经被创建过一次了，可以不需要定义//channel.queueDeclare("queue1", false, false, false, null);// 6： 定义接受消息的回调Channel finalChannel = channel;finalChannel.basicConsume(queueName, true, (s,delivery)-> {System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));}, (s) -> {});System.out.println(queueName + "：开始接受消息");System.in.read();} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null && connection.isOpen()) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}};public static void main(String[] args) {// 启动三个线程去执行new Thread(runnable, "queue1").start();new Thread(runnable, "queue2").start();new Thread(runnable, "queue3").start();}
}

使用三个线程模拟多个消费者，可以同时拿取多个队列

二.direct模式

图示：

producer.Java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 6： 准备发送消息的内容String message = "你好，马明";String  exchangeName = "direct_exchange";String routingKey1 = "q1";String routingKey2 = "q3";// 7: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称/routingkey// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish(exchangeName, routingKey1, null, message.getBytes());channel.basicPublish(exchangeName, routingKey2, null, message.getBytes());System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

consumer.Java

import com.rabbitmq.client.*;
import java.io.IOException;public class Consumer {private static Runnable runnable = () -> {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");//获取队列的名称final String queueName = Thread.currentThread().getName();Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在，则会创建*  Rabbitmq不允许创建两个相同的队列名称，否则会报错。**  @params1： queue 队列的名称*  @params2： durable 队列是否持久化*  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭*  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。*  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。* */// 这里如果queue已经被创建过一次了，可以不需要定义//channel.queueDeclare("queue1", false, false, false, null);// 6： 定义接受消息的回调Channel finalChannel = channel;finalChannel.basicConsume(queueName, true, new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));}}, new CancelCallback() {@Overridepublic void handle(String s) throws IOException {}});System.out.println(queueName + "：开始接受消息");System.in.read();} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null && connection.isOpen()) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}};public static void main(String[] args) {// 启动三个线程去执行new Thread(runnable, "queue1").start();new Thread(runnable, "queue2").start();new Thread(runnable, "queue3").start();}
}

结果消费截图：

三.Topic模式

图示：

producer.java

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 6： 准备发送消息的内容String message = "你好，马明";String  exchangeName = "topics_exchange";String routingKey1 = "name.user";//都可以收到 queue-1 queue-3String routingKey2 = "name";//都可以收到 queue-2// 7: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称/routingkey// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish(exchangeName, routingKey1, null, message.getBytes());channel.basicPublish(exchangeName, routingKey2, null, message.getBytes());System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

consumer.java

import com.rabbitmq.client.*;
import java.io.IOException;public class Consumer {private static Runnable runnable = () -> {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");//获取队列的名称final String queueName = Thread.currentThread().getName();Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 5: 申明队列queue存储消息/**  如果队列不存在，则会创建*  Rabbitmq不允许创建两个相同的队列名称，否则会报错。**  @params1： queue 队列的名称*  @params2： durable 队列是否持久化*  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭*  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。*  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。* */// 这里如果queue已经被创建过一次了，可以不需要定义//channel.queueDeclare("queue1", false, false, false, null);// 6： 定义接受消息的回调Channel finalChannel = channel;finalChannel.basicConsume(queueName, true, new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));}}, new CancelCallback() {@Overridepublic void handle(String s) throws IOException {}});System.out.println(queueName + "：开始接受消息");System.in.read();} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null && connection.isOpen()) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}};public static void main(String[] args) {// 启动三个线程去执行new Thread(runnable, "queue1").start();new Thread(runnable, "queue2").start();new Thread(runnable, "queue3").start();}
}

结果消费截图：

四.使用代码声明交换机和队列

在 RabbitMQ 中，使用 Java 代码声明交换机和队列通常通过 RabbitMQ 提供的客户端库 amqp-client 来实现。以下是一个完整的示例，展示如何使用 Java 代码声明交换机（Exchange）和队列（Queue），并将其绑定在一起。

Direct Exchange：消息通过路由键精确匹配到队列。
Fanout Exchange：广播消息到所有绑定的队列，忽略路由键。
Topic Exchange：支持通配符的路由键匹配。
Headers Exchange：根据消息头（Header）进行匹配，而不是路由键

在生产者中声明：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {public static void main(String[] args) {// 1: 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();// 2: 设置连接属性connectionFactory.setHost("8.137.76.12");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setUsername("admin");connectionFactory.setPassword("admin");Connection connection = null;Channel channel = null;try {// 3: 从连接工厂中获取连接connection = connectionFactory.newConnection("生产者");// 4: 从连接中获取通道channelchannel = connection.createChannel();// 6： 准备发送消息的内容String message = "你好，马明";//需要声明交换机的名字String  exchangeName = "direct_message_exchange";//需要声明交换机的模式String exchangeMode = "direct";//声明交换机 交换机的名称  交换机的模式 是否持久化channel.exchangeDeclare(exchangeName,exchangeMode,true);//声明队列  队列名称  是否持久化 是否具有排他性 时候自动删除  是否携带参数（headers模式需要）channel.queueDeclare("que5",true,false,false,null);channel.queueDeclare("que6",true,false,false,null);//绑定队列到交换机 队列名称  交换机名称 routingKeychannel.queueBind("que5",exchangeName,"order");channel.queueBind("que6",exchangeName,"message");String routingKey1 = "order";//都可以收到 queue-1 queue-3String routingKey2 = "message";//都可以收到 queue-2// 7: 发送消息给中间件rabbitmq-server// @params1: 交换机exchange// @params2: 队列名称/routingkey// @params3: 属性配置// @params4: 发送消息的内容channel.basicPublish(exchangeName, routingKey1, null, message.getBytes());channel.basicPublish(exchangeName, routingKey2, null, message.getBytes());System.out.println("消息发送成功!");} catch (Exception ex) {ex.printStackTrace();System.out.println("发送消息出现异常...");} finally {// 7: 释放连接关闭通道if (channel != null && channel.isOpen()) {try {channel.close();} catch (Exception ex) {ex.printStackTrace();}}if (connection != null) {try {connection.close();} catch (Exception ex) {ex.printStackTrace();}}}}
}

核心的声明代码：

//需要声明交换机的名字String  exchangeName = "direct_message_exchange";//需要声明交换机的模式String exchangeMode = "direct";//声明交换机 交换机的名称  交换机的模式 是否持久化channel.exchangeDeclare(exchangeName,exchangeMode,true);//声明队列  队列名称  是否持久化 是否具有排他性 是否自动删除  是否携带参数（headers模式需要）channel.queueDeclare("que5",true,false,false,null);channel.queueDeclare("que6",true,false,false,null);//绑定队列到交换机 队列名称  交换机名称 routingKeychannel.queueBind("que5",exchangeName,"order");channel.queueBind("que6",exchangeName,"message");