1.Windows下安装RabbitMQ的步骤详解+图解(erlang+RabbitMQ)

2.SpringBoot集成RabbitMQ参考文章

1.RabbitMQ介绍

RabbitMQ是实现AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时,消费者无法快速消费,那么需要一个中间层。保存这个数据。

AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然。AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。

RabbitMQ是一个开源的AMQP实现,服务器端用Erlang语言编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

2.AmqpTemplate,RabbitTemplate

Spring AMQP提供了一个发送和接收消息的操作模板类AmqpTemplate。 AmqpTemplate它定义包含了发送和接收消息等的一些基本的操作功能。RabbitTemplate是AmqpTemplate的一个实现。

RabbitTemplate支持消息的确认与返回,为了返回消息,RabbitTemplate 需要设置mandatory 属性为true,并且CachingConnectionFactory 的publisherReturns属性也需要设置为true。返回的消息会根据它注册的RabbitTemplate.ReturnCallback setReturnCallback 回调发送到给客户端,

一个RabbitTemplate仅能支持一个ReturnCallback 。

为了确认Confirms消息, CachingConnectionFactory 的publisherConfirms 属性也需要设置为true,确认的消息会根据它注册的RabbitTemplate.ConfirmCallback setConfirmCallback回调发送到给客户端。一个RabbitTemplate也仅能支持一个ConfirmCallback.

3.SpringBoot集成RabbitMQ

pom

  1. <?xml version=“1.0” encoding=“UTF-8”?>
  2. <project xmlns=“http://maven.apache.org/POM/4.0.0” xmlns:xsi=“http://www.w3.org/2001/XMLSchema-instance”
  3. xsi:schemaLocation=“http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd”>
  4. <modelVersion>4.0.0</modelVersion>
  5. <groupId>com.example</groupId>
  6. <artifactId>demo</artifactId>
  7. <version>0.0.1-SNAPSHOT</version>
  8. <packaging>jar</packaging>
  9. <name>rabbitMQ</name>
  10. <description>Demo project for Spring Boot</description>
  11. <parent>
  12. <groupId>org.springframework.boot</groupId>
  13. <artifactId>spring-boot-starter-parent</artifactId>
  14. <version>2.0.2.RELEASE</version>
  15. <relativePath/> <!– lookup parent from repository –>
  16. </parent>
  17. <properties>
  18. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  19. <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  20. <java.version>1.8</java.version>
  21. </properties>
  22. <dependencies>
  23. <dependency>
  24. <groupId>org.springframework.boot</groupId>
  25. <artifactId>spring-boot-starter-amqp</artifactId>
  26. </dependency>
  27. <dependency>
  28. <groupId>org.springframework.boot</groupId>
  29. <artifactId>spring-boot-starter-web</artifactId>
  30. </dependency>
  31. <dependency>
  32. <groupId>org.springframework.boot</groupId>
  33. <artifactId>spring-boot-starter-test</artifactId>
  34. <scope>test</scope>
  35. </dependency>
  36. </dependencies>
  37. <build>
  38. <plugins>
  39. <plugin>
  40. <groupId>org.springframework.boot</groupId>
  41. <artifactId>spring-boot-maven-plugin</artifactId>
  42. </plugin>
  43. </plugins>
  44. </build>
  45. </project>

 

自动配置信息  这里我开启ACK消息确认

  1. server.port=8083
  2. #服务器配置
  3. spring.application.name=rabbitmq-hello-sending
  4. #rabbitmq连接参数
  5. spring.rabbitmq.host=localhost
  6. spring.rabbitmq.port=5672
  7. spring.rabbitmq.username=linpeng
  8. spring.rabbitmq.password=123456
  9. # 开启发送确认
  10. spring.rabbitmq.publisher-confirms=true
  11. # 开启发送失败退回
  12. spring.rabbitmq.publisher-returns=true
  13. # 开启ACK
  14. spring.rabbitmq.listener.direct.acknowledge-mode=manual
  15. spring.rabbitmq.listener.simple.acknowledge-mode=manual

创建消息队列  队列名:hello 和 helloObj

  1. package com.example.demo;
  2. import com.rabbitmq.client.Channel;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
  5. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  6. import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
  7. import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  10. @Configuration
  11. public class RabbitConfig {
  12. @Bean
  13. public Queue QueueA() {
  14. return new Queue(“hello”);
  15. }
  16. @Bean
  17. public Queue QueueB() {
  18. return new Queue(“helloObj”);
  19. }
  20. /**
  21. * Fanout 就是我们熟悉的广播模式或者订阅模式,给Fanout交换机发送消息,绑定了这个交换机的所有队列都收到这个消息。
  22. * @return
  23. */
  24. @Bean
  25. FanoutExchange fanoutExchange() {
  26. return new FanoutExchange(“ABExchange”);
  27. }
  28. @Bean
  29. Binding bindingExchangeA(Queue QueueA, FanoutExchange fanoutExchange) {
  30. return BindingBuilder.bind(QueueA).to(fanoutExchange);
  31. }
  32. @Bean
  33. Binding bindingExchangeB(Queue QueueB, FanoutExchange fanoutExchange) {
  34. return BindingBuilder.bind(QueueB).to(fanoutExchange);
  35. }
  36. }

消息发送者 Sender  使用 RabbitTemplate  不采用 AmqpTemplate

  1. package com.example.demo;
  2. import org.springframework.amqp.core.AmqpTemplate;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.amqp.rabbit.support.CorrelationData;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Component;
  8. import org.springframework.stereotype.Service;
  9. import java.util.Date;
  10. //RabbitTemplate.ConfirmCallback
  11. @Service
  12. public class HelloSender implements RabbitTemplate.ReturnCallback {
  13. @Autowired
  14. // private AmqpTemplate rabbitTemplate;
  15. private RabbitTemplate rabbitTemplate;
  16. public void send() {
  17. String context = “你好现在是 ” + new Date() +””;
  18. System.out.println(“HelloSender发送内容 : ” + context);
  19. // this.rabbitTemplate.setConfirmCallback(this);
  20. this.rabbitTemplate.setReturnCallback(this);
  21. this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
  22. if (!ack) {
  23. System.out.println(“HelloSender消息发送失败” + cause + correlationData.toString());
  24. } else {
  25. System.out.println(“HelloSender 消息发送成功 “);
  26. }
  27. });
  28. this.rabbitTemplate.convertAndSend(“hello”, context);
  29. }
  30. public void sendObj() {
  31. MessageObj obj = new MessageObj();
  32. obj.setACK(false);
  33. obj.setId(123);
  34. obj.setName(“zhangsan”);
  35. obj.setValue(“data”);
  36. System.out.println(“发送 : ” + obj);
  37. this.rabbitTemplate.convertAndSend(“helloObj”, obj);
  38. }
  39. @Override
  40. public void returnedMessage(Message message, int i, String s, String s1, String s2) {
  41. System.out.println(“sender return success” + message.toString()+”===”+i+”===”+s1+”===”+s2);
  42. }
  43. // @Override
  44. // public void confirm(CorrelationData correlationData, boolean b, String s) {
  45. // System.out.println(“sender success”);
  46. // }
  47. }

 

消息接受者 Receiver 注解方式接受消息

  1. package com.example.demo;
  2. import com.rabbitmq.client.Channel;
  3. import org.springframework.amqp.core.Message;
  4. import org.springframework.amqp.core.Queue;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
  8. import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
  9. import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
  10. import org.springframework.amqp.support.AmqpHeaders;
  11. import org.springframework.beans.factory.annotation.Configurable;
  12. import org.springframework.context.annotation.Bean;
  13. import org.springframework.messaging.handler.annotation.Headers;
  14. import org.springframework.stereotype.Component;
  15. import java.io.IOException;
  16. import java.util.Date;
  17. import java.util.Map;
  18. @Component
  19. @RabbitListener(queues = “hello”)
  20. public class HelloReceiver {
  21. @RabbitHandler
  22. public void process(String hello,Channel channel, Message message) throws IOException {
  23. System.out.println(“HelloReceiver收到 : ” + hello +”收到时间”+new Date());
  24. try {
  25. //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
  26. channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  27. System.out.println(“receiver success”);
  28. } catch (IOException e) {
  29. e.printStackTrace();
  30. //丢弃这条消息
  31. //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
  32. System.out.println(“receiver fail”);
  33. }
  34. }
  35. }

备注:我们用注解的方式来接受消息 就不要用 自己创建对象实现ChannelAwareMessageListener的方式来接受消息 这种方式还要去全局里面配置 麻烦,直接用@RabbitListener(queues = “hello”)最简单

消息确认  因为我在属性配置文件里面开启了ACK确认 所以如果代码没有执行ACK确认 你在RabbitMQ的后台会看到消息会一直留在队列里面未消费掉 只要程序一启动开始接受该队列消息的时候 又会收到

 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

//消息的标识,false只确认当前一个消息收到,true确认所有consumer获得的消息

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

//ack返回false,并重新回到队列,api里面解释得很清楚

channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);

//拒绝消息

channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);

TestController测试

  1. @Autowired
  2. private HelloSender helloSender;
  3. /**
  4. * 单生产者-单个消费者
  5. */
  6. @RequestMapping(“/test”)
  7. public void hello() throws Exception {
  8. helloSender.send();
  9. }

RabbitMQ后台 两个队列

发送消息

ACK场景测试

我们把HelloReceiver的ACK确认代码注释掉 那消息就算程序收到了 但是未确认ACK导致消息服务器以为他是未成功消费的 后续还会再发

重启程序

——————— 本文来自 屎壳郎情调 的CSDN 博客 ,全文地址请点击:https://blog.csdn.net/linpeng_1/article/details/80505828?utm_source=copy

互联网金融系列-支付清算体系例子-下(理解日切)

笔者上一篇《互联网金融系列-支付清算体系介绍-上》已经比较全面的介绍了以银联为例子的支付清算体系,为了更好的理解里面的运作,本章以两个例子为重点,全...

阅读全文

互联网金融系列-支付清算体系介绍-上

一,支付清算体系的简介 支付清算体系是一个国家的金融基础设施,或说公共服务。我国由央行主管此事,目前大体维持“结算-清算”二级制的支付体系。通俗地讲,...

阅读全文

JDK8 – 当尝试使用Maven生成javadoc时,错误“javax.interceptor.InterceptorBinding未找到的类文件”javadoc plugin

我使用JDK8(尝试它在我的Eclipse工作区与Win x64 u25 JDK在Linux上由Jenkins发布 – jdk-8u20-linux-x64,两个相同的问题)。我有多模块Maven项目(我从包装类...

阅读全文

欢迎留言