定制化开发Spring Boot异步消息之AMQP讲解及实战（附源码）

定制化开发觉得有帮助请点赞关注收藏~~~

AMQP（定制化开发高级消定制化开发息队列协议）定制化开发是一个提供统一消息服定制化开发务的应用层标准高级消息队列协议。是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可 传递消息，并不受客户端/中间件的不同产品，不同开发语言等条件的限制。

下面实现主要用RabbitMQ讲解AMQP实例，因此需要事先安装RabbitMQ和erlang语言

erlang下载地址 https://www.erlang.org/downloads

RabbitMQ下载地址 https://www.rabbitmq.com/download.html 

使用RabbitMQ实现发布/订阅异步消息模式

1：创建发布者应用ch8_2Sender

2：在pom.xml文件中添加依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2.  
  3. -<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0">
  4.  
  5. <modelVersion>4.0.0</modelVersion>
  6.  
  7.  
  8. -<parent>
  9.  
  10. <groupId>org.springframework.boot</groupId>
  11.  
  12. <artifactId>spring-boot-starter-parent</artifactId>
  13.  
  14. <version>2.1.8.RELEASE</version>
  15.  
  16. <relativePath/>
  17.  
  18. <!-- lookup parent from repository -->
  19.  
  20.  
  21. </parent>
  22.  
  23. <groupId>com.ch</groupId>
  24.  
  25. <artifactId>ch8_2Sender</artifactId>
  26.  
  27. <version>0.0.1-SNAPSHOT</version>
  28.  
  29. <name>ch8_2Sender</name>
  30.  
  31. <description>Demo project for Spring Boot</description>
  32.  
  33.  
  34. +<properties>
  35.  
  36.  
  37. -<dependencies>
  38.  
  39.  
  40. -<dependency>
  41.  
  42. <groupId>org.springframework.boot</groupId>
  43.  
  44. <artifactId>spring-boot-starter-amqp</artifactId>
  45.  
  46. </dependency>
  47.  
  48.  
  49. +<dependency>
  50.  
  51.  
  52. -<dependency>
  53.  
  54. <groupId>org.springframework.boot</groupId>
  55.  
  56. <artifactId>spring-boot-starter-test</artifactId>
  57.  
  58. <scope>test</scope>
  59.  
  60. </dependency>
  61.  
  62.  
  63. -<dependency>
  64.  
  65. <groupId>org.springframework.amqp</groupId>
  66.  
  67. <artifactId>spring-rabbit-test</artifactId>
  68.  
  69. <scope>test</scope>
  70.  
  71. </dependency>
  72.  
  73. </dependencies>
  74.  
  75.  
  76. -<build>
  77.  
  78.  
  79. -<plugins>
  80.  
  81.  
  82. -<plugin>
  83.  
  84. <groupId>org.springframework.boot</groupId>
  85.  
  86. <artifactId>spring-boot-maven-plugin</artifactId>
  87.  
  88. </plugin>
  89.  
  90. </plugins>
  91.  
  92. </build>
  93.  
  94. </project>

 3：创建Weather实体类

  1. package com.ch.ch8_2Sender.entity;
  2. import java.io.Serializable;
  3. public class Weather implements Serializable{
  4. 	private static final long serialVersionUID = -8221467966772683998L;
  5. 	private String id;
  6. 	private String city;
  7. 	private String weatherDetail;
  8. 	public String getCity() {
  9. 		return city;
  10. 	}
  11. 	public void setCity(String city) {
  12. 		this.city = city;
  13. 	}
  14. 	public String getWeatherDetail() {
  15. 		return weatherDetail;
  16. 	}
  17. 	public void setWeatherDetail(String weatherDetail) {
  18. 		this.weatherDetail = weatherDetail;
  19. 	}
  20. 	public String getId() {
  21. 		return id;
  22. 	}
  23. 	public void setId(String id) {
  24. 		this.id = id;
  25. 	}
  26. 	@Override
  27. 	public String toString() {
  28. 		return "Weather [id=" + id + ", city=" + city + ", weatherDetail=" + weatherDetail + "]";
  29. 	}
  30. }

4：重写Ch82SenderApplication主类

  1. package com.ch.ch8_2Sender;
  2. import org.springframework.amqp.core.Message;
  3. import org.springframework.amqp.core.MessageBuilder;
  4. import org.springframework.amqp.core.MessageDeliveryMode;
  5. import org.springframework.amqp.rabbit.connection.CorrelationData;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.boot.CommandLineRunner;
  10. import org.springframework.boot.SpringApplication;
  11. import org.springframework.boot.autoconfigure.SpringBootApplication;
  12.  
  13. import com.ch.ch8_2Sender.entity.Weather;
  14. import com.fasterxml.jackson.databind.ObjectMapper;
  15. @SpringBootApplication
  16. public class Ch82SenderApplication implements CommandLineRunner{ 
  17. 	@Autowired
  18.     private ObjectMapper objectMapper;
  19. 	@Autowired 
  20. 	RabbitTemplate rabbitTemplate;
  21. 	public static void main(String[] args) {
  22. 		SpringApplication.run(Ch82SenderApplication.class, args);
  23. 	}
  24. 	/**
  25. 	 * 定义发布者
  26. 	 */
  27. 	@Override
  28. 	public void run(String... args) throws Exception {
  29. 		//定义消息对象
  30. 		Weather weather = new Weather();
  31. 		weather.setId("010");
  32. 		weather.setCity("北京");
  33. 		weather.setWeatherDetail("今天晴到多云，南风5-6级，温度19-26°C");
  34. 		//指定Json转换器，Jackson2JsonMessageConverter默认将消息转换成byte[]类型的消息
  35. 		rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
  36. 		//objectMapper将weather对象转换为JSON字节数组
  37. 		Message msg=MessageBuilder.withBody(objectMapper.writeValueAsBytes(weather))
  38. 				.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
  39.                 .build();
  40.         // 消息唯一ID
  41.         CorrelationData correlationData = new CorrelationData(weather.getId());
  42. 		//使用已封装好的convertAndSend(String exchange , String routingKey , Object message, CorrelationData correlationData)
  43. 		//将特定的路由key发送消息到指定的交换机
  44. 		rabbitTemplate.convertAndSend(
  45. 				"weather-exchange", //分发消息的交换机名称
  46. 				"weather.message", //用来匹配消息的路由Key
  47. 				msg, //消息体
  48. 				correlationData);
  49. 	}
  50. }

5：创建订阅者应用ch8_2Receiver-1

  1. package com.ch.ch8_2Receiver1;
  2. import org.springframework.amqp.rabbit.annotation.Exchange;
  3. import org.springframework.amqp.rabbit.annotation.Queue;
  4. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.messaging.handler.annotation.Payload;
  9. import org.springframework.stereotype.Component;
  10. import com.fasterxml.jackson.databind.ObjectMapper;
  11. /**
  12.  * 定义订阅者Receiver1
  13.  */
  14. @Component
  15. public class Receiver1 {
  16. 	@Autowired
  17.     private ObjectMapper objectMapper;
  18. 	@RabbitListener(
  19. 			bindings = 
  20. 			@QueueBinding(
  21. 				//队列名weather-queue1保证和别的订阅者不一样
  22. 				value = @Queue(value = "weather-queue1",durable = "true"),
  23. 				//weather-exchange与发布者的交换机名相同
  24. 				exchange = @Exchange(value = "weather-exchange",durable = "true",type = "topic"),
  25. 				//weather.message与发布者的消息的路由Key相同
  26. 				key = "weather.message"
  27. 			)	
  28.     )
  29.     @RabbitHandler
  30.     public void receiveWeather(@Payload byte[] weatherMessage)throws Exception{
  31.         System.out.println("-----------订阅者Receiver1接收到消息--------");
  32.         //将JSON字节数组转换为Weather对象
  33.         Weather w=objectMapper.readValue(weatherMessage, Weather.class);
  34.         System.out.println("Receiver1收到的消息内容："+w);
  35.     }
  36. }

6:创建订阅者应用ch8_2Receiver-2

  1. package com.ch.ch8_2Receiver1;
  2. import org.springframework.amqp.rabbit.annotation.Exchange;
  3. import org.springframework.amqp.rabbit.annotation.Queue;
  4. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.messaging.handler.annotation.Payload;
  9. import org.springframework.stereotype.Component;
  10. import com.fasterxml.jackson.databind.ObjectMapper;
  11. /**
  12.  * 定义订阅者Receiver2
  13.  */
  14. @Component
  15. public class Receiver2 {
  16. 	@Autowired
  17.     private ObjectMapper objectMapper;
  18. 	@RabbitListener(
  19. 			bindings = 
  20. 			@QueueBinding(
  21. 				//队列名weather-queue2保证和别的订阅者不一样
  22. 				value = @Queue(value = "weather-queue2",durable = "true"),
  23. 				//weather-exchange与发布者的交换机名相同
  24. 				exchange = @Exchange(value = "weather-exchange",durable = "true",type = "topic"),
  25. 				//weather.message与发布者的消息的路由Key相同
  26. 				key = "weather.message"
  27. 			)	
  28.     )
  29.     @RabbitHandler
  30.     public void receiveWeather(@Payload byte[] weatherMessage)throws Exception{
  31.         System.out.println("-----------订阅者Receiver2接收到消息--------");
  32.         Weather w=objectMapper.readValue(weatherMessage, Weather.class);
  33.         //将JSON字节数组转换为Weather对象
  34.         System.out.println("Receiver2收到的消息内容："+w);
  35.      
  36.     }
  37. }

接下来分别运行发布者和订阅者的主类即可，发现一个发布者发布的消息可以被多个订阅者订阅。

网站建设定制开发 软件系统开发定制 定制软件开发 软件开发定制 定制app开发 app开发定制 app开发定制公司 电商商城定制开发 定制小程序开发 定制开发小程序 客户管理系统开发定制 定制网站 定制开发 crm开发定制 开发公司 小程序开发定制 定制软件 收款定制开发 企业网站定制开发 定制化开发 android系统定制开发 定制小程序开发费用 定制设计 专注app软件定制开发 软件开发定制定制 知名网站建设定制 软件定制开发供应商 应用系统定制开发 软件系统定制开发 企业管理系统定制开发 系统定制开发