消息生产rabbitmq-provider
pom.xml添加
|
1 2 3 4 5 6 7 8 |
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> |
application.yml 添加
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
spring: #给项目来个名字 application: name: rabbitmq-provider #配置rabbitMq 服务器 rabbitmq: host: 192.168.3.137 port: 5672 username: root password: root #虚拟host 可以不设置,使用server默认host virtual-host: / #消息确认配置项 #确认消息已发送到交换机(Exchange) publisher-confirm-type: correlated #确认消息已发送到队列(Queue) publisher-returns: true |
Service代码
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
@Slf4j @Configuration public class RabbitConfig { @Bean public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(); rabbitTemplate.setConnectionFactory(connectionFactory); //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数 rabbitTemplate.setMandatory(true); rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info("**ConfirmCallback: "+"相关数据："+correlationData); log.info("**ConfirmCallback: "+"确认情况："+ack); log.info("**ConfirmCallback: "+"原因："+cause); } }); rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() { @Override public void returnedMessage(ReturnedMessage message) { log.info("^^ReturnCallback: "+"消息："+message); } }); return rabbitTemplate; } } |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
/** * 直连型交换机 - 生产者 */ @Slf4j @Configuration public class AlertDirectRabbitConfig { //队列 起名：TestDirectQueue @Bean public Queue AlertDirectQueue() { log.info("Alert直连型队列AlertDirectQueue"); // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效 // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。 // return new Queue("AlertDirectQueue",true,true,false); //一般设置一下队列的持久化就好,其余两个就是默认false return new Queue("AlertDirectQueue",true); } //Direct交换机 起名：AlertDirectExchange @Bean DirectExchange AlertDirectExchange() { log.info("Alert直连型交换机AlertDirectExchange"); // return new DirectExchange("AlertDirectExchange",true,true); return new DirectExchange("AlertDirectExchange",true,false); } //绑定 将队列和交换机绑定, 并设置用于匹配键：AlertDirectRouting @Bean Binding bindingDirect() { log.info("Alert直连型队列绑定到上交换机，主键为AlertDirectRouting"); return BindingBuilder.bind(AlertDirectQueue()).to(AlertDirectExchange()).with("AlertDirectRouting"); } } |
Controller 代码
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
@RestController @Slf4j public class SendAlertController { @Autowired RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法 @GetMapping("/sendAlertMessage") public String sendAlertMessage() { String messageId = String.valueOf(UUID.randomUUID()); String messageData = "alert message, hello!"; String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); Map<String,Object> map=new HashMap<>(); map.put("messageId",messageId); map.put("messageData",messageData); map.put("createTime",createTime); //将消息携带绑定键值：AlertDirectRouting 发送到交换机AlertDirectExchange rabbitTemplate.convertAndSend("AlertDirectExchange", "AlertDirectRouting", map); return "ok"; } } |
消费消息rabbitmq-consumer
pom.xml添加
|
1 2 3 4 5 6 7 8 |
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> |
application.yml 添加
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
spring: #给项目来个名字 application: name: rabbitmq-consumer #配置rabbitMq 服务器 rabbitmq: host: 192.168.3.137 port: 5672 username: root password: root #虚拟host 可以不设置,使用server默认host virtual-host: / #消息确认配置项 #确认消息已发送到交换机(Exchange) publisher-confirm-type: correlated #确认消息已发送到队列(Queue) publisher-returns: true |
Service代码
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
@Configuration public class MessageListenerConfig { @Autowired private CachingConnectionFactory connectionFactory; @Autowired private AlertAckReceiver alertAckReceiver; @Bean public SimpleMessageListenerContainer simpleMessageListenerContainer() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setConcurrentConsumers(1); container.setMaxConcurrentConsumers(1); // RabbitMQ默认是自动确认，这里改为手动确认消息 container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置一个队列 container.setQueueNames("AlertDirectQueue"); //如果同时设置多个如下： 前提是队列都是必须已经创建存在的 // container.setQueueNames("TestDirectQueue","fanout.A"); //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues //container.setQueues(new Queue("TestDirectQueue",true)); //container.addQueues(new Queue("TestDirectQueue2",true)); //container.addQueues(new Queue("TestDirectQueue3",true)); container.setMessageListener(alertAckReceiver); return container; } } |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
@Slf4j @Component public class AlertAckReceiver implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理 String msg = message.toString(); //可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据 String[] msgArray = msg.split("'"); Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3); String messageId=msgMap.get("messageId"); String messageData=msgMap.get("messageData"); String createTime=msgMap.get("createTime"); log.info(" AlertAckReceiver messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); log.info("消费的主题消息来自："+message.getMessageProperties().getConsumerQueue()); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息 channel.basicAck(deliveryTag, true); //第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝 // channel.basicReject(deliveryTag, true); if ("AlertDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){ log.info("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue()); log.info("消息成功消费到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); log.info("执行AlertDirectQueue中的消息的业务处理流程......"); } if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){ log.info("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue()); log.info("消息成功消费到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); log.info("执行fanout.A中的消息的业务处理流程......"); } log.info(".................................................................."); } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } private Map<String, String> mapStringToMap(String str, int entryNum) { log.info(str); str = str.substring(1, str.length() - 1); String[] strs = str.split(",",entryNum); Map<String, String> map = new HashMap<String, String>(); for (String string : strs) { String key = string.split("=")[0].trim(); String value = string.split("=")[1]; map.put(key, value); } return map; } } |