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

Java 如何实现一个简单 RabbitMQ 示例

news 来源：原创 2024/9/20 20:48:45

本文建立在你在 Linux 上完成安装 RabbitMQ 的基础上。

1、生产者代码

顾名思义，生产者是用来生产消息供消费者消费

package com.wen.rabbitmq;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.HashMap;public class Product {public final static String QUEUE_NAME = "my-mq-queue";public final static String HOST = "10.106.182.54";public final static String USERNAME = "root";public final static String PASSWORD = "123456";public final static String EXCHANGE_NAME = "";public static void main(String[] args) {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);try {Connection connection = factory.newConnection();Channel channel = connection.createChannel();HashMap<String, Object> map = new HashMap<>();map.put("x-message-ttl", 10000);channel.queueDeclare(QUEUE_NAME, false, false, true, map);String message = "Hello, How are you";channel.basicPublish(EXCHANGE_NAME, QUEUE_NAME, null, message.getBytes());System.out.println("消息已经发出！");channel.close();connection.close();} catch (IOException e) {throw new RuntimeException(e);}}
}

2、消费者代码

package com.wen.rabbitmq;import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.HashMap;public class Consumer {public final static String QUEUE_NAME = "my-mq-queue";public final static String HOST = "10.106.182.54";public final static String USERNAME = "root";public final static String PASSWORD = "123456";public final static String EXCHANGE_NAME = "";public static void main(String[] args) {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);Connection connection;try {connection = factory.newConnection();Channel channel = connection.createChannel();HashMap<String, Object> map = new HashMap<>();map.put("x-message-ttl", 10000);channel.queueDeclare(QUEUE_NAME, false, false ,true, map);DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)throws IOException {String s = new String(body, "UTF-8");System.out.println("消费端接收到消息：" + s);}};channel.basicConsume(QUEUE_NAME, true, consumer);} catch (Exception e) {e.printStackTrace();}}
}

3、示例解读

1、首先，我们先创建 QUEUE_NAME 队列为消息的发送做准备，RabbitMQ 的消息是需要存储在消息队列中的，消息队列是测试 QPS 的一个关键点。HOST USERNAME PASSWORD 等没啥好说的，我们要链接到自己的 RabbitMQ 服务上，使用 ConnectionFactory 进行连接。

2、消息生产出来后，需要一个通道（channel）去传输信息，这个通道会和交换器有交集，这里作为一个简单demo就不再多讲。

3、channel.queueDeclare(QUEUE_NAME, false, false ,true, map);

queueDeclare方法是用来创建一个队列，生产者和消费者都能使用这个方法进行创建队列，但如果消费者在同一个信道上订阅了另一个队列，就无法再声明队列了。必须要先取消订阅。并且，消费者和生产者创建的队列里面的信息要一致。

参数一：队列名字
参数二：是否设置持久化
参数三：是否设置排他
参数四：是否设置自动删除
参数五：设置队列中的一些参数，这是一个 Map，常见的有过期时间，优先级等。

注：如果生产者设置了过期时间，那么消费者创建的队列中也要有相应的参数。

4、channel.basicPublish(EXCHANGE_NAME, QUEUE_NAME, null, message.getBytes());，这串代码是用来发送消息。

参数一：交换器名字
参数二：路由键名字，交换器会根据路由键将消息存储到相应的队列中。
参数三：消息的基本属性集，内包含14个成员，可自行查找。
参数四：消息体，是一个 byte[] 。

5、消费者这部分也要先建立一个消息队列 my-mq-queue ，和生产者保持一致。消费消息有两种模式，推模式和拉模式。本文示例采用的推模式，推模式使用 basicConsume 方法进行消费。接收消息通过实现 Consumer 接口或者继承 DefaultConsumer 类来实现。

6、DefaultConsumer 类中我们要重写 handleDelivery 方法，这个方法是用来处理消息传递的细节。handleDelivery 方法的参数如下：