package com.mes.common;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.rabbitmq.client.*;
|
|
import java.util.*;
|
import java.util.concurrent.ArrayBlockingQueue;
|
import java.util.concurrent.BlockingQueue;
|
|
public class RabbitMQUtils {
|
// ObjectMapper对象，用于序列化和反序列化JSON
|
private static ObjectMapper objectMapper = new ObjectMapper();
|
private static String host = "10.153.19.150"; // RabbitMQ 主机名
|
private static Map<String, Object> args = new HashMap<>(); // 队列参数
|
|
static {
|
// 设置队列参数
|
args.put("x-max-length-bytes", 1024 * 1024);
|
}
|
|
// private static Map<String, Object> args = new HashMap<>();
|
// args.put("x-max-length-bytes",1024 * 1024);
|
// 发送消息到 RabbitMQ 队列中
|
public String sendMessage(String message, String queueName) throws Exception {
|
// 创建连接工厂并设置主机名
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
// 使用 try-with-resources 语句创建连接和通道，并发送消息
|
try (Connection connection = factory.newConnection();
|
Channel channel = connection.createChannel()) {
|
// 声明队列
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
// 发布消息到队列
|
channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
|
System.out.println(" [x] Sent '" + message + "' to queue '" + queueName + "'");
|
}
|
|
return message;
|
}
|
|
|
// 从 RabbitMQ 队列中接收消息
|
public String receiveMessage(String queueName) throws Exception {
|
// 创建连接工厂并设置主机名
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
// 创建阻塞队列
|
BlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(1);
|
|
// 使用 try-with-resources 语句创建连接和通道，并接收消息
|
try (Connection connection = factory.newConnection();
|
Channel channel = connection.createChannel()) {
|
// 声明队列
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
// 设置消息接收回调
|
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
|
String receivedMessage = new String(delivery.getBody(), "UTF-8");
|
// System.out.println(" [x] Received '" + receivedMessage + "'");
|
|
// 将接收到的消息放入阻塞队列
|
try {
|
messageQueue.put(receivedMessage);
|
} catch (InterruptedException e) {
|
e.printStackTrace();
|
}
|
};
|
|
// 消费队列中的消息
|
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
|
|
});
|
|
// 阻塞并等待获取消息
|
return messageQueue.take();
|
}
|
}
|
|
|
//消费消息
|
public List<String> readMessages(String queueName,boolean is) throws Exception {
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
List<String> messages = new ArrayList<>();
|
try (Connection connection = factory.newConnection();
|
Channel channel = connection.createChannel()) {
|
boolean autoAck = false;
|
|
GetResponse response = channel.basicGet(queueName, autoAck);
|
if (response != null) {
|
String message = new String(response.getBody(), "UTF-8");
|
messages.add(message);
|
// 手动确认消息处理完成
|
if(is){
|
long deliveryTag = response.getEnvelope().getDeliveryTag();
|
channel.basicAck(deliveryTag, false);
|
}
|
|
}
|
}
|
|
return messages;
|
}
|
|
|
|
private static Set<String> sentMessageIds = new HashSet<>();
|
//根据id发送消息
|
public static boolean sendMessageWithId(String queueName, String message, String messageId) throws Exception {
|
if (sentMessageIds.contains(messageId)) {
|
System.err.println("Message with ID " + messageId + " has already been sent.");
|
return false; // 消息重复，发送失败
|
}
|
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
|
.messageId(messageId)
|
.build();
|
|
channel.basicPublish("", queueName, properties, message.getBytes("UTF-8"));
|
System.out.println("Sent message with ID: " + messageId);
|
|
sentMessageIds.add(messageId); // 将 messageId 添加到已发送集合中
|
|
return true; // 消息成功发送
|
} catch (Exception e) {
|
System.err.println("Failed to send message: " + e.getMessage());
|
return false; // 消息发送失败
|
}
|
}
|
//根据id消费消息
|
public static String consumeMessageById( String messageId,String queueName) throws Exception {
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
GetResponse response;
|
while ((response = channel.basicGet(queueName, false)) != null) {
|
String receivedMessage = new String(response.getBody(), "UTF-8");
|
if (response.getProps().getMessageId().equals(messageId)) {
|
long deliveryTag = response.getEnvelope().getDeliveryTag();
|
channel.basicAck(deliveryTag, false);
|
System.out.println("Selected message: "+messageId+ receivedMessage);
|
return receivedMessage; // 返回选定的消息内容
|
} else {
|
// 对于不符合条件的消息，进行 Nack 操作
|
// long deliveryTag = response.getEnvelope().getDeliveryTag();
|
// channel.basicNack(deliveryTag, false, true);
|
return "Specified message not found in the queue.";
|
}
|
}
|
|
return "Specified message not found in the queue.";
|
}
|
}
|
|
//消费指定消息
|
public static String consumeSelectedMessage(int messageToConsume, String queueName) throws Exception {
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
for (int i = 0; i < messageToConsume; i++) {
|
GetResponse response = channel.basicGet(queueName, false);
|
if (response == null) {
|
return "Queue does not have enough messages.";
|
}
|
// long deliveryTag = response.getEnvelope().getDeliveryTag();
|
// channel.basicAck(deliveryTag, false);
|
}
|
|
GetResponse selectedResponse = channel.basicGet(queueName, false);
|
if (selectedResponse != null) {
|
byte[] body = selectedResponse.getBody();
|
String selectedMessage = new String(body, "UTF-8");
|
long deliveryTag = selectedResponse.getEnvelope().getDeliveryTag();
|
channel.basicAck(deliveryTag, false);
|
return selectedMessage;
|
} else {
|
return "Specified message not found in the queue.";
|
}
|
}
|
}
|
|
// 发送 JSON 消息到队列
|
public static boolean sendJsonMessage(Object message, String queueName) throws Exception {
|
// 创建连接工厂并设置主机名
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
// 使用 try-with-resources 语句创建连接和通道，并发送 JSON 消息
|
try (Connection connection = factory.newConnection();
|
Channel channel = connection.createChannel()) {
|
// 声明队列
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
// 将对象转换为 JSON 字符串
|
String jsonMessage = objectMapper.writeValueAsString(message);
|
|
// 发布 JSON 消息到队列
|
channel.basicPublish("", queueName, null, jsonMessage.getBytes());
|
System.out.println(" [x] Sent JSON message: '" + jsonMessage + "' to queue '" + queueName + "'");
|
|
return true; // 发送消息成功
|
} catch (Exception e) {
|
e.printStackTrace();
|
return false; // 发送消息失败
|
}
|
}
|
|
// 接收 JSON 消息并转换为对象
|
public static <T> T receiveJsonMessage(Class<T> valueType, String queueName) throws Exception {
|
// 创建连接工厂并设置主机名
|
ConnectionFactory factory = new ConnectionFactory();
|
factory.setHost(host);
|
|
// 使用 try-with-resources 语句创建连接和通道，并接收 JSON 消息
|
try (Connection connection = factory.newConnection();
|
Channel channel = connection.createChannel()) {
|
// 声明队列
|
channel.queueDeclare(queueName, false, false, false, args);
|
|
// 获取队列中的消息
|
GetResponse response = channel.basicGet(queueName, true);
|
if (response != null) {
|
byte[] body = response.getBody();
|
String jsonMessage = new String(body, "UTF-8");
|
System.out.println(" [x] Received JSON message: '" + jsonMessage + "' from queue '" + queueName + "'");
|
|
// 将 JSON 消息转换为对象
|
return objectMapper.readValue(jsonMessage, valueType);
|
} else {
|
return null;
|
}
|
} catch (Exception e) {
|
e.printStackTrace();
|
return null; // 接收消息失败
|
}
|
}
|
|
}
|
|
// RabbitMQUtils.sendMessage("Hello, world!", "module_queue");
|
//RabbitMQUtils.receiveMessage("module_queue");
|
//RabbitMQUtils.sendJsonMessage(someObject, "another_queue");
|
//RabbitMQUtils.receiveJsonMessage(SomeClass.class, "another_queue");
|
// 需要将 someObject 替换为你要发送的对象，并将 SomeClass 替换为你要接收并转换的对象类型。
|
