• java springBoot实现RabbitMq消息队列 生产者，消费者

    1.RabbitMq的数据源配置文件

    1. # 数据源配置
    2. spring:
    3.     rabbitmq:
    4.         host: 127.0.0.1
    5.         port: 5672
    6.         username: root
    7.         password: root
    8.         #消息发送和接收确认
    9.         publisher-confirms: true
    10.         publisher-returns: true
    11.         listener:
    12.             direct:
    13.                 acknowledge-mode: manual
    14.             simple:
    15.                 acknowledge-mode: manual
    16.                 retry:
    17.                     enabled: true #是否开启消费者重试
    18.                     max-attempts: 5 #最大重试次数
    19.                     initial-interval: 2000 #重试间隔时间（单位毫秒）

    2.maven依赖

    

    org.springframework.boot
    spring-boot-starter-amqp

    3.RabbitMq文件目录预览

    4. RabbitMq的Action文件

    1. package com.zq.cnz.mq.constant;
    2.  
    3. public enum Action {
    4. 	ACCEPT, // 处理成功
    5. 	RETRY, // 可以重试的错误
    6. 	REJECT, // 无需重试的错误
    7. }

    5.RabbitMq的QueueContent文件

    1. package com.zq.cnz.mq.constant;
    2.  
    3. /**
    4.  * @ClassName: QueueContent
    5.  * @Description: 消息队列名称
    6.  * @author 吴顺杰
    7.  * @date 2023年11月15日
    8.  *
    9.  */
    10. public class QueueContent {
    11. 	/**
    12. 	 * 测试消息队列
    13. 	 */
    14. 	public static final String TEST_MQ_QUEUE = "test_mq_queue";
    15.  
    16.  
    17. 	/**
    18. 	 * 测试消息队列交换机
    19. 	 */
    20. 	public static final String TEST_MQ_QUEUE_EXCHANGE = "test_mq_queue_exchange";
    21.  
    22. 	/**
    23. 	 * 测试消息延迟消费队列
    24. 	 */
    25. 	public static final String TEST_MQ_QUEUE_TIME_DELAY_EXCHANGE = "test_mq_queue_time_delay_exchange";
    26.  
    27. }

    6.消息队列生产者MessageProvider方法

    1. package com.zq.cnz.mq;
    2.  
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.zq.common.utils.IdUtils;
    5. import org.slf4j.Logger;
    6. import org.slf4j.LoggerFactory;
    7. import org.springframework.amqp.AmqpException;
    8. import org.springframework.amqp.core.Message;
    9. import org.springframework.amqp.core.MessagePostProcessor;
    10. import org.springframework.amqp.rabbit.connection.CorrelationData;
    11. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    12. import org.springframework.stereotype.Component;
    13.  
    14. /**
    15.  * 消息队列生产
    16.  */
    17. @Component
    18. public class MessageProvider implements RabbitTemplate.ConfirmCallback {
    19. 	static Logger logger = LoggerFactory.getLogger(MessageProvider.class);
    20.  
    21. 	/**
    22. 	 * RabbitMQ 模版消息实现类
    23. 	 */
    24. 	protected RabbitTemplate rabbitTemplate;
    25.  
    26. 	public MessageProvider(RabbitTemplate rabbitTemplate) {
    27. 		this.rabbitTemplate = rabbitTemplate;
    28. 		this.rabbitTemplate.setMandatory(true);
    29. 		this.rabbitTemplate.setConfirmCallback(this);
    30. 	}
    31.  
    32. 	private String msgPojoStr;
    33.  
    34. 	/**
    35. 	 * 推送消息至消息队列
    36. 	 *
    37. 	 * @param msg
    38. 	 * @param queueName
    39. 	 */
    40. 	public void sendMqMessage(String queueName,String msg) {
    41. 		try {
    42. 			JSONObject object = JSONObject.parseObject(msg);
    43. 			String msgId = IdUtils.fastUUID().toString();
    44. 			object.put("msgId", msgId);
    45. 			msg = object.toString();
    46. 			msgPojoStr = msg;
    47. 			logger.info("推送消息至" + queueName + "消息队列,消息内容" + msg);
    48. 			rabbitTemplate.convertAndSend(queueName, msg);
    49. 		} catch (AmqpException e) {
    50. 			e.printStackTrace();
    51. 			logger.error("推送消息至消息队列异常 ,msg=" + msg + ",queueName=" + queueName, e);
    52. 		}
    53. 	}
    54.  
    55. 	/**
    56. 	 * 推送广播消息
    57. 	 *
    58. 	 * @param exchangeName
    59. 	 * @param msg
    60. 	 */
    61. 	public void sendFanoutMsg(String exchangeName, String msg) {
    62. 		try {
    63. 			JSONObject object = JSONObject.parseObject(msg);
    64. 			String msgId = IdUtils.fastUUID().toString();
    65. 			object.put("msgId", msgId);
    66. 			msg = object.toString();
    67. 			msgPojoStr = msg;
    68. 			logger.info("推送广播消息至交换机" + exchangeName + ",消息内容" + msg);
    69. 			rabbitTemplate.convertAndSend(exchangeName, "", msg);
    70. 		} catch (AmqpException e) {
    71. 			e.printStackTrace();
    72. 			logger.error("推送广播至交换机异常 ,msg=" + msg + ",exchangeName=" + exchangeName, e);
    73. 		}
    74. 	}
    75.  
    76. 	/**
    77. 	 * 发送延时消息
    78. 	 *
    79. 	 * @param queueName
    80. 	 * @param msg
    81. 	 */
    82. 	public void sendTimeDelayMsg(String queueName, String exchangeName, String msg, Integer time) {
    83. 		try {
    84. 			JSONObject object = JSONObject.parseObject(msg);
    85. 			String msgId = IdUtils.fastUUID().toString();
    86. 			object.put("msgId", msgId);
    87. 			msg = object.toString();
    88. 			msgPojoStr = msg;
    89. 			logger.info("推送延时消息至" + exchangeName + "," + queueName + "消息队列,消息内容" + msg + ",延时时间" + time + "秒");
    90. 			rabbitTemplate.convertAndSend(exchangeName, queueName, msg, new MessagePostProcessor() {
    91. 				@Override
    92. 				public Message postProcessMessage(Message message) throws AmqpException {
    93. 					message.getMessageProperties().setHeader("x-delay", time * 1000);
    94. 					return message;
    95. 				}
    96. 			});
    97. 		} catch (AmqpException e) {
    98. 			e.printStackTrace();
    99. 			logger.error("推送消息至消息队列异常 ,msg=" + msg + ",exchangeName=" + exchangeName + ",queueName=" + queueName
    100. 					+ ",time=" + time, e);
    101. 		}
    102. 	}
    103.  
    104. 	@Override
    105. 	public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    106. 		if (ack) {
    107. 			logger.info(msgPojoStr + ":消息发送成功");
    108. 		} else {
    109. 			logger.warn(msgPojoStr + ":消息发送失败:" + cause);
    110. 		}
    111. 	}
    112.  
    113. }

    7.消息队列消费者RabbitMqConfiguration文件配置

    1. package com.zq.cnz.mq;
    2.  
    3. import com.zq.cnz.mq.constant.QueueContent;
    4. import org.springframework.amqp.core.*;
    5. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    6. import org.springframework.amqp.rabbit.core.RabbitAdmin;
    7. import org.springframework.beans.factory.annotation.Autowired;
    8. import org.springframework.context.annotation.Bean;
    9. import org.springframework.context.annotation.Configuration;
    10.  
    11. import javax.annotation.Resource;
    12. import java.util.HashMap;
    13. import java.util.Map;
    14.  
    15. @Configuration
    16. public class RabbitMqConfiguration {
    17.  
    18. 	@Resource
    19. 	RabbitAdmin rabbitAdmin;
    20.  
    21.  
    22. 	// 创建初始化RabbitAdmin对象
    23. 	@Bean
    24. 	public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
    25. 		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
    26. 		// 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
    27. 		rabbitAdmin.setAutoStartup(true);
    28. 		return rabbitAdmin;
    29. 	}
    30.  
    31. 	/**
    32. 	 * 测试消息队列
    33. 	 *
    34. 	 * @return
    35. 	 */
    36. 	@Bean
    37. 	public Queue TEST_QUEUE() {
    38. 		return new Queue(QueueContent.TEST_MQ_QUEUE);
    39. 	}
    40.  
    41. 	/**
    42. 	 * 测试交换机
    43. 	 *
    44. 	 * @return
    45. 	 */
    46. 	@Bean
    47. 	FanoutExchange TEST_MQ_QUEUE_EXCHANGE() {
    48. 		return new FanoutExchange(QueueContent.TEST_MQ_QUEUE_EXCHANGE);
    49. 	}
    50.  
    51.  
    52. 	/**
    53. 	 * 测试延迟消费交换机
    54. 	 *
    55. 	 * @return
    56. 	 */
    57. 	@Bean
    58. 	public CustomExchange TEST_MQ_QUEUE_TIME_DELAY_EXCHANGE() {
    59. 		Map<String, Object> args = new HashMap<>();
    60. 		args.put("x-delayed-type", "direct");
    61. 		return new CustomExchange(QueueContent.TEST_MQ_QUEUE_TIME_DELAY_EXCHANGE, "x-delayed-message", true, false, args);
    62. 	}
    63.  
    64. 	/**
    65. 	 * 测试延迟消费交换机绑定延迟消费队列
    66. 	 *
    67. 	 * @return
    68. 	 */
    69. 	@Bean
    70. 	public Binding banTestQueue() {
    71. 		return BindingBuilder.bind(TEST_QUEUE()).to(TEST_MQ_QUEUE_TIME_DELAY_EXCHANGE()).with(QueueContent.TEST_MQ_QUEUE).noargs();
    72. 	}
    73.  
    74.  
    75. 	// 创建交换机和对列,跟上面的Bean的定义保持一致
    76. 	@Bean
    77. 	public void createExchangeQueue() {
    78. 		//测试消费队列
    79. 		rabbitAdmin.declareQueue(TEST_QUEUE());
    80. 		//测试消费交换机
    81. 		rabbitAdmin.declareExchange(TEST_MQ_QUEUE_EXCHANGE());
    82. 		//测试延迟消费交换机
    83. 		rabbitAdmin.declareExchange(TEST_MQ_QUEUE_TIME_DELAY_EXCHANGE());
    84. 	}
    85.  
    86. }

    8.TestQueueConsumer 消息队列消费+延迟消费

    1. package com.zq.cnz.mq.MessageConsumer;
    2.  
    3. import com.alibaba.druid.util.StringUtils;
    4. import com.alibaba.fastjson.JSONObject;
    5. import com.rabbitmq.client.Channel;
    6. import com.zq.cnz.mq.MessageProvider;
    7. import com.zq.cnz.mq.constant.Action;
    8. import com.zq.cnz.mq.constant.QueueContent;
    9. import com.zq.common.utils.IdUtils;
    10. import com.zq.common.utils.RedisUtils;
    11. import com.zq.common.utils.spring.SpringUtils;
    12. import org.slf4j.Logger;
    13. import org.slf4j.LoggerFactory;
    14. import org.springframework.amqp.core.Message;
    15. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    16. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    17. import org.springframework.beans.factory.annotation.Autowired;
    18. import org.springframework.stereotype.Component;
    19.  
    20. import java.io.IOException;
    21.  
    22. /**
    23.  * 测试消息队列消费
    24.  */
    25. @Component
    26. @RabbitListener(queues = QueueContent.TEST_MQ_QUEUE)
    27. public class TestQueueConsumer {
    28.     @Autowired
    29.     private RedisUtils redisUtils;
    30.     static final Logger logger = LoggerFactory.getLogger(TestQueueConsumer.class);
    31.  
    32.     @RabbitHandler
    33.     public void handler(String msg, Channel channel, Message message) throws IOException {
    34.         if (!StringUtils.isEmpty(msg)) {
    35.             JSONObject jsonMsg = JSONObject.parseObject(msg);
    36. //            logger.info("TestQueueConsumer:"+jsonMsg.toJSONString());
    37.             Action action = Action.RETRY;
    38. //			获取消息ID
    39.             String msgId = jsonMsg.getString("msgId");
    40. //			消费次数+1
    41.             redisUtils.incr("MQ_MSGID:" + msgId, 1);
    42.             redisUtils.expire("MQ_MSGID:" + msgId, 60);
    43.             try {
    44.                 logger.info("测试消费队列消费成功啦，消费信息："+jsonMsg.getString("test"));
    45.                 action = Action.ACCEPT;
    46.             } catch (Exception e) {
    47.                 logger.error("MQ_MSGID：" + msgId + ",站控权限请求关闭接口异常,msg=" + msg, e);
    48.             } finally {
    49.                 // 通过finally块来保证Ack/Nack会且只会执行一次
    50.                 if (action == Action.ACCEPT) {
    51.                     channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    52.                 } else if (action == Action.RETRY) {
    53. //					判断当前消息消费次数,已经消费3次则放弃消费
    54.                     if ((Integer) redisUtils.get("MQ_MSGID:" + msgId) >= 3) {
    55.                         logger.error("MQ_MSGID：" + msgId + ",异步处理超出失败次数限制,msg=" + msg);
    56.                         channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    57.                     } else {
    58. //						回归队列重新消费
    59.                         channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    60.                     }
    61.                 } else {
    62.                     channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    63.                 }
    64.             }
    65.         }
    66.     }
    67. }

    9.TestExchangeConsumer 交换机广播模式 

    1. package com.zq.cnz.mq.MessageConsumer;
    2.  
    3. import com.alibaba.fastjson.JSON;
    4. import com.alibaba.fastjson.JSONArray;
    5. import com.alibaba.fastjson.JSONObject;
    6. import com.alibaba.fastjson.serializer.SerializerFeature;
    7. import com.rabbitmq.client.Channel;
    8. import com.zq.cnz.mq.constant.Action;
    9. import com.zq.cnz.mq.constant.QueueContent;
    10. import com.zq.common.utils.IdUtils;
    11. import com.zq.common.utils.RedisUtils;
    12. import com.zq.common.utils.StringUtils;
    13. import com.zq.common.utils.spring.SpringUtils;
    14. import org.slf4j.Logger;
    15. import org.slf4j.LoggerFactory;
    16. import org.springframework.amqp.core.ExchangeTypes;
    17. import org.springframework.amqp.core.Message;
    18. import org.springframework.amqp.rabbit.annotation.*;
    19. import org.springframework.beans.factory.annotation.Autowired;
    20. import org.springframework.stereotype.Component;
    21.  
    22. import javax.annotation.Resource;
    23. import java.io.IOException;
    24. import java.util.List;
    25.  
    26. /**
    27.  * 测试交换机消费
    28.  */
    29. @Component
    30. @RabbitListener(bindings = @QueueBinding(value = @Queue(), exchange = @Exchange(value = QueueContent.TEST_MQ_QUEUE_EXCHANGE, type = ExchangeTypes.FANOUT)))
    31. public class TestExchangeConsumer {
    32.     static final Logger logger = LoggerFactory.getLogger(TestExchangeConsumer.class);
    33.     @Resource
    34.     private RedisUtils redisUtils;
    35.  
    36.     @RabbitHandler
    37.     public void handler(String msg, Channel channel, Message message) throws IOException {
    38.         if (!StringUtils.isEmpty(msg)) {
    39. //            logger.info("接收交换机生产者消息:{}", msg);
    40.             Action action = Action.ACCEPT;
    41.             // 请求参数
    42.             JSONObject jsonMsg = JSONObject.parseObject(msg);
    43. //			获取消息ID
    44.             String msgId = jsonMsg.getString("msgId");
    45.  
    46. //			消费次数+1
    47.             redisUtils.incr("MQ_MSGID:" + msgId, 1);
    48.             redisUtils.expire("MQ_MSGID:" + msgId, 60);
    49.             try {
    50.  
    51.                 Integer CMD = jsonMsg.getInteger("cmd");
    52.                 if (CMD==1) {
    53.                     logger.info("cmd1测试消费队列消费成功啦，消费信息："+jsonMsg.getString("test"));
    54.                 }else if(CMD==2){
    55.                     logger.info("cmd2测试消费队列消费成功啦，消费信息："+jsonMsg.getString("test"));
    56.                 }
    57.                 action = Action.ACCEPT;
    58.             } catch (Exception e) {
    59.                 action = Action.REJECT;
    60.                 e.printStackTrace();
    61.             } finally {
    62.                 // 通过finally块来保证Ack/Nack会且只会执行一次
    63.                 if (action == Action.ACCEPT) {
    64.                     channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    65.  
    66.                 } else if (action == Action.RETRY) {
    67. //					判断当前消息消费次数,已经消费3次则放弃消费
    68.                     if ((Integer) redisUtils.get("MQ_MSGID:" + msgId) >= 3) {
    69.                         logger.error("MQ_MSGID:：" + msgId + ",换电失败消息队列消费了三次,msg=" + msg);
    70.                         channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    71.                     } else {
    72. //						回归队列重新消费
    73.                         channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    74.                     }
    75.                 } else {
    76.                     channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    77.                 }
    78.  
    79.             }
    80.         }
    81.     }
    82. }

    运行项目 调用RabbitmqTestController生产RabbitMq消息体, TestExchangeConsumer和TestQueueConsumer自动消费

    1. package com.zq.web.controller.tool;
    2. import com.alibaba.fastjson.JSONObject;
    3. import com.zq.cnz.mq.MessageProvider;
    4. import com.zq.cnz.mq.constant.QueueContent;
    5. import com.zq.common.utils.IdUtils;
    6. import org.springframework.web.bind.annotation.GetMapping;
    7. import org.springframework.web.bind.annotation.RequestMapping;
    8. import org.springframework.web.bind.annotation.RestController;
    9. import javax.annotation.Resource;
    10.  
    11.  
    12. /**
    13.  * 消息队列测试
    14.  */
    15. @RestController
    16. @RequestMapping("/test/mq")
    17. public class RabbitmqTestController {
    18.     @Resource
    19.     private MessageProvider messageProvider;
    20.  
    21.     /**
    22.      * 查询储能站信息列表
    23.      */
    24.     @GetMapping("/putMq")
    25.     public void putMq(){
    26.         JSONObject obj=new JSONObject();
    27.         obj.put("test","测试数据");
    28.         //推送消息至消息队列
    29.        messageProvider.sendMqMessage(QueueContent.TEST_MQ_QUEUE,obj.toString());
    30.         obj.put("cmd",1);
    31.         obj.put("test","这是广播消费");
    32.         //推送广播消息
    33.         messageProvider.sendFanoutMsg(QueueContent.TEST_MQ_QUEUE_EXCHANGE,obj.toString());
    34.         //发送延时消息
    35.         obj.put("cmd",2);
    36.         obj.put("test","这是延迟消费");
    37.         messageProvider.sendTimeDelayMsg(QueueContent.TEST_MQ_QUEUE,QueueContent.TEST_MQ_QUEUE_TIME_DELAY_EXCHANGE,obj.toString(),2*60);
    38.  
    39.     }
    40. }

  • 相关阅读:
    adb and 软件架构笔记
    肖sir__设计测试用例方法之_（白盒测试）
    Prometheus Operator 通过additional 添加target
    时间序列预测：用电量预测 04 Std_Linear（多元线性回归算法 & 数据标准化）
    BurpSuite官方实验室之逻辑漏洞
    突发技术故障对工作进程的影响及其应对策略——以电脑硬盘损坏为例
    java学习day17（Java核心类库）IO流
    【ORACLE】Oracle里有“time”数据类型吗？--关于对Oracle数据类型的一点研究
    javase----java基础面试题01-05
    电脑监控软件：保护企业核心信息资产，防止数据泄露
  • 原文地址：https://blog.csdn.net/qq_37557563/article/details/134422240