RabbitMQ 最新版安装，设置，java接入使用（详细教程） - ToB企服应用市场:ToB评测及商务社交产业平台

rpm -ivh erlang-26.2.5.2-1.el7.x86_64.rpm

复制代码

rpm -ivh rabbitmq-server-3.13.6-1.el8.noarch.rpm

复制代码

rabbitmq-diagnostics status

复制代码

vi /etc/rabbitmq/rabbitmq.config

复制代码

[
{rabbit, [{tcp_listeners, [{"0.0.0.0", 51091}]}]},
{rabbitmq_management, [
{listener, [{port,59876}, {ssl, false}]}
]}
].

复制代码

#启动
systemctl start rabbitmq-server
#停止关闭
systemctl stop rabbitmq-server
#重启
systemctl restart rabbitmq-server
#开机启动
systemctl enable rabbitmq-server
#查看状态
systemctl status rabbitmq-server

复制代码

rabbitmq-plugins enable rabbitmq_management

复制代码

#新增人员
rabbitmqctl add_user hua abc123uuPP
#设置权限
rabbitmqctl set_permissions -p / hua ".*" ".*" ".*"
#设置为管理员
rabbitmqctl set_user_tags hua administrator

复制代码

<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.20.0</version>
</dependency>

复制代码

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.util.concurrent.TimeoutException;
/**
* @author hua
* @date 2024-08-21 18:01
*/
public class TestRabbitMQ {
private final static String QUEUE_NAME = "hello";
public static void main1(String[] args) {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("xx.xx.xx.xx");
factory.setPort(51091);
factory.setUsername("java_producer");
factory.setPassword("java_producer");
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
} catch (TimeoutException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
public static void main(String[] argv) throws Exception {
// 创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("xx.xx.xx.xx");
factory.setPort(51091);
factory.setUsername("java_consumer");
factory.setPassword("java_consumer");
// 连接到 RabbitMQ 服务器
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
// 声明队列（确保队列存在）
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
// 定义回调函数，当有消息送达时执行
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
};
// 消费消息
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
}
}
}

复制代码

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author hua
* @date 2024-08-22
*/
@Service
public class RabbitMqServiceImpl {
private static final Logger logger = LogManager.getLogger(RabbitMqServiceImpl.class);
private static final String QUEUE_NAME = "test";
private Connection connection;
private Channel channel;
public RabbitMqServiceImpl() {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("xx.xx.xx.xx");
factory.setPort(51091);
factory.setUsername("java_producer");
factory.setPassword("java_producer");
//如果不指定虚拟机默认会使用/
factory.setVirtualHost("test");
try {
this.connection = factory.newConnection();
this.channel = connection.createChannel();
this.channel.queueDeclare(QUEUE_NAME, false, false, false, null);
logger.info("RabbitMqServiceImpl initialized successfully.");
} catch (IOException | TimeoutException e) {
e.printStackTrace();
logger.error("Failed to initialize RabbitMqServiceImpl: {}", e.getMessage());
throw new RuntimeException("Failed to initialize RabbitMqServiceImpl", e);
}
}
public void sendMessage(String message) {
try {
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
} catch (IOException e) {
e.printStackTrace();
logger.error("Failed to send message: {}", e.getMessage());
}
}
public void close() {
try {
if (channel != null && channel.isOpen()) {
channel.close();
}
if (connection != null && connection.isOpen()) {
connection.close();
}
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
}
}

复制代码

import com.qyhua.common.table.db_hex_fail_log.entity.DbHexFailLog;
import com.qyhua.common.table.db_hex_fail_log.service.impl.DbHexFailLogServiceImpl;
import com.rabbitmq.client.*;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;
/**
* @author hua
* @date 2024-08-22
*/
@Service
public class RabbitMqServiceImpl {
private static final Logger logger = LogManager.getLogger(RabbitMqServiceImpl.class);
private static final String QUEUE_NAME = "hex_kyc";
private Connection connection;
private Channel channel;
//存放所有消息，确认时删除，没确认的保存到数据库
private ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
@Autowired
DbHexFailLogServiceImpl dbHexFailLogService;
@PostConstruct
public void init() {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("xx.xx.xx.xx");
factory.setPort(xxx);
factory.setUsername("java_producer");
factory.setPassword("java_producer");
factory.setVirtualHost("xxxx");
factory.setConnectionTimeout(3000);
try {
this.connection = factory.newConnection();
this.channel = connection.createChannel();
this.channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 启用发布者确认模式
this.channel.confirmSelect();
setupConfirmListener();
logger.info("RabbitMqServiceImpl initialized successfully.");
} catch (IOException | TimeoutException e) {
logger.error("Failed to initialize RabbitMqServiceImpl: {}", e.getMessage(), e);
throw new RuntimeException("Failed to initialize RabbitMqServiceImpl", e);
}
}
public void sendMessage(String message) {
try {
long nextSeqNo = channel.getNextPublishSeqNo();
outstandingConfirms.put(nextSeqNo, message);
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
logger.info(" [x] Sent '{}'", message);
} catch (Exception e) {
logger.error("Failed to send message: {}", e.getMessage(), e);
saveFailedMessageToDatabase(message,"CF");
}
}
// 设置接收监听器，记录未确认的消息
private void setupConfirmListener() {
ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
if (multiple) {
outstandingConfirms.headMap(deliveryTag + 1).clear();
} else {
outstandingConfirms.remove(deliveryTag);
}
System.out.println("Message confirmed ok deliveryTag="+deliveryTag);
};
ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
if (multiple) {
// 获取从起点到 `deliveryTag + 1` 之间的所有未确认的消息
ConcurrentNavigableMap<Long, String> unconfirmedMessages = outstandingConfirms.headMap(deliveryTag + 1);
List<String> FailList= new ArrayList<>();
for (Map.Entry<Long, String> entry : unconfirmedMessages.entrySet()) {
String failedMessage = entry.getValue();
logger.error("Message not confirmed: deliveryTag={}, message={}", entry.getKey(), failedMessage);
FailList.add(failedMessage);
}
saveFailedMessageToDatabaseBy(FailList); // 批量保存到数据库
unconfirmedMessages.clear(); // 清除这些未确认的消息
} else {
String failedMessage = outstandingConfirms.get(deliveryTag);
logger.error("Message not confirmed: deliveryTag={}, message={}", deliveryTag, failedMessage);
saveFailedMessageToDatabase(failedMessage,"SF");
outstandingConfirms.remove(deliveryTag); // 移除单条未确认的消息
}
};
channel.addConfirmListener(ackCallback, nackCallback);
}
private void saveFailedMessageToDatabaseBy(List<String> failList) {
List<DbHexFailLog> list=new ArrayList<>(failList.size());
LocalDateTime now = LocalDateTime.now();
for (String message : failList) {
DbHexFailLog f=new DbHexFailLog();
f.setInHexStr(message);
f.setCtime(now);
f.setFlag("SF");
list.add(f);
}
dbHexFailLogService.saveBatch(list,list.size());
failList.clear();
}
private void saveFailedMessageToDatabase(String message,String flag) {
DbHexFailLog f=new DbHexFailLog();
f.setInHexStr(message);
f.setCtime(LocalDateTime.now());
f.setFlag(flag);
dbHexFailLogService.save(f);
}
@PreDestroy
public void close() {
try {
if (channel != null && channel.isOpen()) {
channel.close();
}
if (connection != null && connection.isOpen()) {
connection.close();
}
logger.info("RabbitMqServiceImpl resources closed successfully.");
} catch (IOException | TimeoutException e) {
logger.error("Failed to close RabbitMqServiceImpl resources: {}", e.getMessage(), e);
}
}
}

复制代码

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>

复制代码

Spring:
rabbitmq:
host: xx.xx.xx.xx
port: 51091
username: java_consumer
password: java_consumer
virtual-host: hellow
connection-timeout: 6000

复制代码

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @author hua
* @date 2024-08-22
*/
@Component
public class MessageProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private Queue queue;
public void sendMessage(String message) {
rabbitTemplate.convertAndSend(queue.getName(), message);
System.out.println(" [x] Sent '" + message + "'");
}
}

复制代码

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @author hua
* @date 2024-08-22
*/
@Component
public class RabbitListener {
private static final Logger logger = LogManager.getLogger(RabbitListener.class);
@RabbitListener(queues = "test")
public void receiveMessage(String message) {
try {
System.out.println("rabbit rev <- "+message);
//具体业务
} catch (Exception e) {
e.printStackTrace();
logger.error("rabbit err= ", e);
}
}
}

复制代码