RabbitMQ--SpringBoot

RabbitMQ--SpringBoot

1、消息队列

消息队列的功能：
    1、异步

    2、削峰

    3、解耦
    
消息队列的规范：
	1、JMS
		Java MessageService,实际上是指JMS API。JMS是由Sun公司早期提出的消息标准，旨在为java应用提供统一的消息操作，包括create、send、receive等。JMS已经成为Java Enterprise Edition的一部分。从使用角度看，JMS和JDBC担任差不多的角色,相关产品有	activeMq、Redis等。
	
	2、AMQP
		AMQP（advanced message queuing protocol）,顾名思义，AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。意味着我们可以使用Java的AMQP provider，同时使用一个python的producer加一个rubby的consumer。从这一点看，AQMP可以用http来进行类比，不关心实现的语言，只要大家都按照相应的数据格式去发送报文请求，不同语言的client均可以和不同语言的server链接。具体实现有rabbitMQ.

2、SpringBoot整合rabbitMq

rabbitmq是AMQP的实现，加入amqp的starter即可。

<!--        rabbitmq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

<!--楼上starter包含了这两个依赖-->
<dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-messaging</artifactId>
      <version>5.2.8.RELEASE</version>
      <scope>compile</scope>
    </dependency>
    <dependency>
      <groupId>org.springframework.amqp</groupId>
      <artifactId>spring-rabbit</artifactId>
      <version>2.2.10.RELEASE</version>
      <scope>compile</scope>
    </dependency>

在yml中配置rabbitmq的相关参数。

spring:
    rabbitmq:
        host: xxxx	#所有参数都有默认值，不指定则是localhost
        port: 5672  #默认通信地址
        #默认虚拟主机等
        username: guest
        password: guest #默认

关于rabbitmq的配置类，在这里用代码生成了交换机、队列，并且通过路由进行了绑定。

如果rabbitmq的服务上已经有了这些，也可以直接用。

/**
 * @author cgl
 * @version 1.0
 * @date 2020/9/16 16:51
 */
@Configuration
@EnableRabbit
public class MqConfig {

    //设置rabbitTemplate的序列化方式，不设置默认使用jdk序列化
    //jdk序列化，对象需要继承Serializable
    //解析收到的消息为对象时，使用的是setter,所以对象需要有无参构造器
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //使用json序列化
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }


    //交换机
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("ek.directExchange");
    }
    //队列
    @Bean
    public Queue queue(){
        return new Queue("ek.queue",true);
    }

    //绑定两者
    @Bean
    public Binding binding(){
        return BindingBuilder.bind(queue()).to(directExchange()).with("ekRouting");
    }


}

测试一下：

@Data
@AllArgsConstructor
@NoArgsConstructor //rabbitMq需要空参构造器
public class User implements Serializable {//默认序列化需要

    private String name;
    private Integer id;
    private String sex;
    private Integer age;
}

@SpringBootTest
@RunWith(SpringRunner.class)
public class MqTest {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    public void test(){
        //转换并发送，队列接受的是Json数据
    	rabbitTemplate.convertAndSend("ek.directExchange","ekRouting","Hello RabbitMq!");
        User user = new User("Jack", 10010, "男", 18);
        rabbitTemplate.convertAndSend("ek.directExchange","ekRouting",user);

        //接收队列中的一个消息，返回对应的Json数据，然后被转换为对象
        //System.out.println(rabbitTemplate.receiveAndConvert("ek.queue"));
    }


    @Test
    public void test2(){
        User user = new User("Jack", 10010, "男", 18);

        //使用字节数组作为消息体，消息头使用默认即可
        Message message=new Message((user.toString()).getBytes(),new MessageProperties());
        
        //直接发送，需要一个Message类型的参数，参数中封装了字节数据（消息体），以及传输参数（消息头）
        rabbitTemplate.send("ek.directExchange","ekRouting",message);
        
        //如果接收数据，也是Message对象，自行解析即可。
        {Message receive = rabbitTemplate.receive("ek.queue");
        System.out.println(receive);
        byte[] body = receive.getBody();
        String s = new String(body);
        System.out.println(s);}
    }
}

3、序列化

//jdk默认序列化使用二进制，
/*
content_type:	
	application/x-java-serialized-object
Payload
	218 bytes
	Encoding: base64
*/
rO0ABXNyABRjb20uY2dsLmVrLnBvam8uVXNlcigRewaMt+OQAgAETAADYWdldAATTGphdmEvbGFuZy9JbnRlZ2VyO0wAAmlkcQB+AAFMAARuYW1ldAASTGphdmEvbGFuZy9TdHJpbmc7TAADc2V4cQB+AAJ4cHNyABFqYXZhLmxhbmcuSW50ZWdlchLioKT3gYc4AgABSQAFdmFsdWV4cgAQamF2YS5sYW5nLk51bWJlcoaslR0LlOCLAgAAeHAAAAASc3EAfgAEAAAnGnQABEphY2t0AAPnlLc=
    

    
//使用字节数组
/*
content_type:	
	application/octet-stream
Payload
    47 bytes
    Encoding: string
*/
 User(name=Jack, id=10010, sex=男, age=18)

 //使用json序列化
 /*
 headers:	
    __TypeId__:	com.cgl.ek.pojo.User
content_encoding:	
	UTF-8
content_type:	
	application/json
Payload
    47 bytes
    Encoding: string	
 */
{"name":"Jack","id":10010,"sex":"男","age":18}

可以看到无论使用json还是字节数组都优于jdk的序列化方式

4、消费者、监听

生产的消息会被消费，之前直接接收也是消费。

常用的是直接监听消息队列，有事件进入就直接取出。

Spring中提供了消息监听，只要监听到队列有数据，就直接调用方法。

@Component
public class RabbitConsumer {

    //注解，监听队列
    @RabbitListener(queues = "ek.queue")
    public void receive(String msg) {
	//监听到之后，方法就会执行
        System.out.println(msg);
    }


}

循环发送消息，测试监听器。

@Autowired
    RabbitConsumer rabbitConsumer;

    @Test
    public void  test4() throws InterruptedException {

        //循环发送消息给队列
        for(int i=0;i<5;i++){
            int nextInt = new Random().nextInt(100);
            LocalDateTime now = LocalDateTime.now();
            DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("现在是yyyy-MM-dd hh:mm:ss a");
            String format = dateTimeFormatter.format(now);
            rabbitTemplate.convertAndSend("ek.directExchange","ekRouting", format+nextInt);
            //睡会儿，让监听器得以启动
            Thread.sleep(3000);
        }

        //让主线程多睡会儿，便于观察可视化界面
        Thread.sleep(10000);

    }

此时控制栏会输出数据：
    "现在是2020-09-17 11:22:03 上午66"
    "现在是2020-09-17 11:22:08 上午37"
    "现在是2020-09-17 11:22:13 上午53"
    "现在是2020-09-17 11:22:18 上午11"
    "现在是2020-09-17 11:22:23 上午78"

主线程休眠的时间里，监听器获得足够的时间启动，在可视化界面中可以看到消费者一栏出现ip，点进去，可以看到连接和信道：
    Channel: 180.115.x.x:54576 -> 172.x.x.x:5672 (1)
    Channel: 180.115.x.x:54576 -> 172.x.x.x:5672 (2)
    Channel: 180.115.x.x:54576 -> 172.x.x.x:5672 (3)
    ....