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

RabbitMQ操作实战

news 2025/9/14 3:21:52

1.RabbitMQ安装

RabbitMQ Windows 安装、配置、使用 - 小白教程-腾讯云开发者社区-腾讯云下载erlang：http://www.erlang.org/downloads/https://cloud.tencent.com/developer/article/2192340

Windows 10安装RabbitMQ及延时消息插件rabbitmq_delayed_message_exchange - 民工黑猫 - 博客园安装RabbitMQ服务器第一步：下载erlang 原因：RabbitMQ服务端代码是使用并发式语言Erlang编写的，安装Rabbit MQ的前提是安装Erlang。下载地址：http://www.erlang.org/downloads 第二步：下载RabbitMQ 下载地址：https://https://www.cnblogs.com/yyee/p/14281111.html

2.生产端确保消息发送到交换机和路由键

application.yml

spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestvirtual-host: /publisher-confirm-type: CORRELATED # 交换机的确认publisher-returns: true # 队列的确认
logging:level:com.atguigu.mq.config: info

RabbitConfig

@Configuration
@Slf4j
public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void initRabbitTemplate() {rabbitTemplate.setConfirmCallback(this);rabbitTemplate.setReturnsCallback(this);}// 发送到交换机-成功或失败@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {log.info("confirm() 回调函数打印 correlationData:{}, ack:{}, cause:{}", correlationData, ack, cause);}// 发送到队列-失败@Overridepublic void returnedMessage(ReturnedMessage returned) {log.info("returnedMessage() 回调函数 msg:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",new String(returned.getMessage().getBody()), returned.getReplyCode(), returned.getReplyText(), returned.getExchange(), returned.getRoutingKey());}
}

RabbitMQTest

public static final String EXCHANGE_DIRECT = "exchange.direct.order";public static final String ROUTING_KEY = "order";@Autowiredprivate RabbitTemplate rabbitTemplate;//发消息-生产端确保消息发到交换机和路由键rabbitTemplate.convertAndSend(EXCHANGE_DIRECT + "~", ROUTING_KEY + "~", "Message Test Confirm~~~ ~~~");

3.消费端手动ack

application.yml

spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestvirtual-host: /listener:simple:acknowledge-mode: manual # 把消息确认模式改为手动确认prefetch: 1 # 每次从队列中取回消息的数量

MyMessageListener

public static final String QUEUE_NAME  = "queue.order";//接消息-消费端手动ack//@RabbitListener(queues = {QUEUE_NAME})public void processMessage(String dataString, Message message, Channel channel) throws IOException {//获取消息的deliveryTaglong deliveryTag = message.getMessageProperties().getDeliveryTag();try {//核心操作log.info("消费端 消息内容：" + dataString);System.out.println(10 / 0);//核心操作成功返ACKchannel.basicAck(deliveryTag, false);} catch (Exception e) {//获取消息是否重复投递Boolean redelivered = message.getMessageProperties().getRedelivered();//核心操作失败返NACKif (!redelivered) {//第一次投递,重新放回队列 (requeue：是否重新放回队列)channel.basicNack(deliveryTag, false, true);//channel.basicReject(deliveryTag, true);} else {//重复投递,不重新放回队列 (requeue：是否重新放回队列)channel.basicNack(deliveryTag, false, false);//channel.basicReject(deliveryTag, false);}}}

4.消息设置超时时间（通过消息后置处理器）

rabbitTemplate.convertAndSend(EXCHANGE_TIMEOUT, ROUTING_KEY_TIMEOUT, "Test timeout", message -> {message.getMessageProperties().setExpiration("4000");return message;});

5.延迟消息（延迟插件）

Community Plugins | RabbitMQ RabbitMQ设置延迟消息的交换机

生产端发送延时消息（通过消息后置处理器）

        //延迟消息-延时插件(最多俩天内)String msg = "Test delay message by plugin " + new SimpleDateFormat("HH:mm:ss").format(new Date());rabbitTemplate.convertAndSend(EXCHANGE_DELAY, ROUTING_KEY_DELAY, msg, message -> {//x-delay参数必须基于x-delayed-message-exchange插件才能生效message.getMessageProperties().setHeader("x-delay", "10000");return message;});

消费端消费延时消息

@RabbitListener(queues = {QUEUE_DELAY})public void processMessageDelay(String dataString, Message message, Channel channel) throws IOException {log.info("[delay message][消息本身]" + dataString);log.info("[delay message][当前时间]" + new SimpleDateFormat("HH:mm:ss").format(new Date()));channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}

查看全文

http://www.lryc.cn/news/543863.html