Springboot消息JMSAMQPRabbbitMQSpringboot整合篇13/14/15/16/17/18/19

  应用场景1----异步处理&＃xff1a;

                 写进消息队列只用5ms 然后异步处理。 

2.应用解耦

    3.流量削峰

             队列是定长的&＃xff0c;没进队列的话就返回秒杀失败。

                   消息代理&＃xff1a;就是消息中间件的服务器。消息发到服务器&＃xff0c;服务器把消息发到指定的目的地。

                   通信机制如下图&＃xff1a;对于点对点的只要有人拿到了其他人就不能拿到了。

           4就是出队和入队5就是消息发布者发布到一个主题里面&＃xff0c;bcd监听&＃xff0c;只要一到达就可以收到。

           JMS&＃xff1a;AMQP是规范&＃xff0c;不同

           spring使用的话只要导入starter就可以了&＃xff0c;然后就是自动配置类。

           AMQP是跨平台的有的比较多的。

13------------------------------------------------------------------------------------------------------------------------

        这个是由erlang实现的。

        消息生产者将消息发送到服务器--消息代理&＃xff0c;发给服务器里面的交换机。接收消息&＃xff0c;路由到消息队列。队列和交换器的关联关系是绑定&＃xff0c;就是路由关系表。

            信道&＃xff1a;解决登陆复用的。就是一条tcp连接的多个信道。

消费者和消息队列用tcp复用进行连接。

14-------------------------------------------------------------------------------------------------------------------------

      消息队列和交换器是多对多的关系。

交换器的四种类型。

direct&＃xff1a;直连&＃xff0c;路由件和绑定的一模一样的时候才发送。

fanout&＃xff1a;就是交换器消息抵达fanout交换器的时候&＃xff0c;就是绑定的队列每人发一份的&＃xff0c;就是广播模式。JMS里面发布订阅的参考实现。

topic&＃xff1a;允许对路由件进行模糊匹配有选择的发给队列。类似于正则表达式。

headers&＃xff1a;这个用的比较少的。

总结&＃xff1a;

15--------------------------------------------------------------------------------------------------------------------------------------

         讲解了如何用docker下载镜像。

         搭建RabbitMQ&＃xff1a;

            安装带management的是带管理界面的。

docker pull rabbitmq:3-management

  启动运行镜像&＃xff1a;

docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq 1f53405491bcd是后台运行 -p是暴露端口 两个端口 5672是容器的 15672web端口

    访问下&＃xff1a;http://192.168.244.130:15672/

用户名和密码是guest

03&＃xff1a;44

中间绿色的是四个队列。此外层是绑定规则。

新建exchange

      添加交换器  Durable是是否持久化就是下次再来是不是数据还在。

       新建两个交换机。

   目前有三个交换器&＃xff0c;继续添加消息队列。

   继续添加&＃xff1a;

-----------------------------------------------------------------------------

    交换器绑定队列&＃xff1a;

    点进去交换器&＃xff1a;

     按照以上得方法给其他得交换机也绑定队列。

topic的

      解释下上面第一个的意思 只要是队列的名字是atguigu.XXXXX我的exchange都绑定上。

      绑定主要绑定的是exchange和queue

      三步&＃xff1a;建交换机-队列-绑定队列

------------------------------------------------------------

      给交换器发送消息看哪个队列能收到消息。

      测试下消息&＃xff0c;点击交换器&＃xff1a;

  在队列atguigu里面查看&＃xff1a;

         fanout的路由发送。

         可以看到全部发出了。

       -----------------------------------------------

         最后测试topic&＃xff0c;根据匹配规则发送给队列。

都发了。

应答之后删除掉。选择这个。

  就是不同的类型的exchange绑定queue设置名称&＃xff0c;按照名称和规则绑定。

16--------------------------------------------------------------------------------------------------------------------------

         创建过程整合测试rabbitmq     

  选择web和rabbitmq

org.springframework.bootspring-boot-starter-amqp

   看到这个。

    分析下自动配置的原理&＃xff1a;

我们可以看下这个自动配置类&＃xff0c;都是在自动配置的配置文件里面的。

点进去这个配置文件&＃xff1a;

根据配置文件我们开始配置。    

spring.rabbitmq.host&＃61;192.168.244.130
spring.rabbitmq.username&＃61;guest
spring.rabbitmq.password&＃61;guest
spring.rabbitmq.port&＃61;5672
#spring.rabbitmq.virtual-host&＃61;

基本的配置看自动配置配置的。

---------------------------------------------------------------------------------------------------------------------------------------------------------------------

如何发送消息和接受消息呢&＃xff1f;我们还看自动配置

redisTemplate这个类的作用是给rabbitMq发送和接受消息的。 

这个是系统管理组件&＃xff0c;声明交换器和队列管理什么的。

注入消息组件。

---------------------------------

测试几个消息&＃xff1a;    

/*** 1、单播&＃xff08;点对点&＃xff09;*/&＃64;Testpublic void contextLoads() {//Message需要自己构造一个;定义消息体内容和消息头&＃xff0c;最后一哥参数查看Message类//rabbitTemplate.send(exchage,routeKey,message);//object默认当成消息体&＃xff0c;只需要传入要发送的对象&＃xff0c;自动序列化发送给rabbitmq&＃xff1b;//rabbitTemplate.convertAndSend(exchage,routeKey,object);Map map &＃61; new HashMap<>();map.put("msg","这是第一个消息");map.put("data", Arrays.asList("helloworld",123,true));//对象被默认序列化以后发送出去 参数分别是交换器的名字 路由键 和消息体rabbitTemplate.convertAndSend("exchange.direct","atguigu.news",new Book("西游记","吴承恩"));
}

   发送测试。

 这个会有乱码是因为序列化是默认的java的序列化的方式。

 接受消息&＃xff0c;肯定在队列里面接受消息&＃xff1a;

//接受数据,如何将数据自动的转为json发送出去&＃64;Testpublic void receive(){//参数是队列的名字Object o &＃61; rabbitTemplate.receiveAndConvert("atguigu.news");System.out.println(o.getClass());System.out.println(o);}

    ----------------------------------------------------------------------------------------------------

       注意一点收到消息队列就没了因为就出队了&＃xff0c;不是广播的只保证有一个收到。

         如何将数据自动转为json发送出去&＃xff1a;

         原始的结果&＃xff1a;

       点进去simpleMessageConverter()&＃xff1a;

         我们可以换一个MessageConverter。

         看这个自动配置不为null的话就用我们自己定义的。

package com.atguigu.config;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;&＃64;Configuration
public class MyAMQPConfig {&＃64;Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}
}


在次测试&＃xff1a;

现在存取对象&＃xff1a;

package com.atguigu.bean;public class Book {private String bookName;private String author;public Book() {}public Book(String bookName, String author) {this.bookName &＃61; bookName;this.author &＃61; author;}public String getBookName() {return bookName;}public void setBookName(String bookName) {this.bookName &＃61; bookName;}public String getAuthor() {return author;}public void setAuthor(String author) {this.author &＃61; author;}&＃64;Overridepublic String toString() {return "Book{" &＃43;"bookName&＃61;&＃39;" &＃43; bookName &＃43; &＃39;\&＃39;&＃39; &＃43;", author&＃61;&＃39;" &＃43; author &＃43; &＃39;\&＃39;&＃39; &＃43;&＃39;}&＃39;;}
}


广播&＃xff1a;

/*** 广播*/&＃64;Testpublic void sendMsg(){Map map &＃61; new HashMap<>();map.put("msg","这是第二个消息");map.put("data", Arrays.asList("helloworld",123,true));//参数依次是转换器的名称(广播) 路由件 消息体rabbitTemplate.convertAndSend("exchange.fanout","",map);}

17--------------------------------------------------------------------------------------------------------------------------

实际开发要需要监听场景&＃xff1a;

类似库存系统和订单系统。   

场景&＃xff1a;两个系统订单系统和库存系统。下单的信息放在消息队列里面&＃xff0c;库存系统实时的监听订单系统的消息。

           一旦有新的订单进来库存系统由相应的操作。

package com.atguigu.service;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;&＃64;Service
public class BookService {//监听是监听的下消息队列 直接将消息反序列化为book对象 相当于强转&＃64;RabbitListener(queues &＃61; "atguigu.news")public void receive(String book){System.out.println("收到消息&＃xff1a;"&＃43;book);}&＃64;RabbitListener(queues &＃61; "atguigu")public void receive02(Message message){System.out.println(message.getBody());System.out.println(message.getMessageProperties());}
}


  注意这个使用之前要加上这个&＃xff0c;开启基于注解的RabbitMQ&＃xff1a;

     切记。 

 清空队列了。

测试如下方法&＃xff1a;

1.主程序开启基于注解的RabbltMQ。

2.注解监控消息,一旦在队列加入了值&＃xff0c;直接监控器就会取出来。

18-----------------------------------------------------------------------------------------------------------------------------

管理系统&＃xff1a;创建消息队列什么的。这个的目的是在程序中创建消息队列 交换器 绑定规则 和消息队列

 测试类里面测试&＃xff0c;注入&＃xff1a;

&＃64;AutowiredAmqpAdmin amqpAdmin;

在自动配置类RabbitAutoConfiguration

里面找到这个组件&＃xff1a;

搜索下Exchange&＃xff0c;看下继承关系。

创建交换器&＃xff1a;

&＃64;Testpublic void createExchange(){//创建exchangeamqpAdmin.declareExchange(new DirectExchange("amqpadmin.exchange"));System.out.println("创建完成");amqpAdmin.declareQueue(new Queue("amqpadmin.queue",true));//创建绑定规则//参数&＃xff1a;目的地就是绑定到哪个队列&＃xff0c;绑定的类型是队列还是交换器&＃xff0c;绑定的交换器给哪个交换器绑定&＃xff0c;路由件&＃xff0c;其他是nullamqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE,"amqpadmin.exchange","amqp.haha",null));}

 运行&＃xff1a;http://192.168.244.130:15672/

创建队列&＃xff1a;      

&＃64;Testpublic void createExchange(){//amqpAdmin.declareExchange(new DirectExchange("amqpadmin.exchange"));//System.out.println("创建完成");//创建队列amqpAdmin.declareQueue(new Queue("amqpadmin.queue",true));//创建绑定规则//amqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE,"amqpadmin.exchange","amqp.haha",null));//amqpAdmin.de}

   创建绑定队则&＃xff1a;

&＃64;Testpublic void createExchange(){//amqpAdmin.declareExchange(new DirectExchange("amqpadmin.exchange"));//System.out.println("创建完成");//amqpAdmin.declareQueue(new Queue("amqpadmin.queue",true));//创建绑定规则//参数&＃xff1a;目的地就是绑定到哪个队列&＃xff0c;绑定的类型是队列还是交换器&＃xff0c;
绑定的交换器给哪个交换器绑定&＃xff0c;路由件&＃xff0c;其他是nullamqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE,"amqpadmin.exchange","amqp.haha",null));//amqpAdmin.delete......}

19-----------------------------------------------------------------------------------------------------------------------------

github地址&＃xff1a;https://github.com/FandyWw/spring-boot-02-amqp