13 Spring boot message rabbitmq

1，安装
测试服务器centos 6.7 ，分别下载erlang-20.3.8.21-1.el6.x86_64.rpm，rabbitmq-server-3.7.14-1.el6.noarch.rpm，socat-1.7.3.2-1.1.el6.x86_64.rpm对应版本，使用rpm命令进行安装。
2，安装完成后，执行如下命令：

chkconfig rabbitmq-server on
service rabbit-server start

启动rabbitmq。如果出现启动异常，crash的情况，确认是否存在端口占用的情况。
3，默认会创建一个名为guest密码为guest的用户，只用在本地通过localhost进行连接，如果需要在其他服务器上连接需要进行access control设置。创建测试用户

rabbitmqctl add_user rabbitmq 12345678
rabbitmqctl set_permissions rabbitmq ".*" ".*" ".*"
rabbitmqctl set_user_tags rabbitmq administrator
rabbitmqctl delete_user guest

AMQP是一个用来处理发布者和消费者的消息协议。publishe发布消息，consumer选取消息并处理，broker的作用就是保证从发布者发布的消息能到达消费者。broker的关键组件是exchanges和queues（图片来源https://spring.io/blog/2010/06/14/understanding-amqp-the-protocol-used-by-rabbitmq/）。

如图所示，发布者发布消息到指定的exchanges，然后消费者从queue中拉取消息或queue向消费者推送消息。发布者和消费者通过exchange发现彼此。exchange的创建根据实际情况，要么由发布者创建要么由消费者创建。
消息从exchange到queue的路由过程，queue要依附于一个指定的exchange。一般情况下，消费者创建一个queue同时依附于一个exchange。由exchange接收的消息会匹配到对应的queue，这个过程叫做binding。
AMQP的消息由三部分组成，headers,properties，byte[] data。其中headers和properties是key-value形式的数据。消息体就是字节数组，如果发送文本信息需要指定编码。
headers中有一个routing-key的属性，broker用这个属性将消息匹配到指定的queue，每个queue也会有一个binding key，如果routing-key和binding key匹配，则queue就会接受这个消息。
exchange有四种类型：direct（binding key和routing-key全匹配），topic（类似于direct，但是可以使用wildcards），fanout（忽略binding key 和routing-key），headers。
基于以上对于rabbitmq的理解来使用spring实现简单的发送消息和消费消息。
1，添加起步配置

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

2，

@SpringBootApplication
@EnableScheduling
@EnableRabbit
public class DemoForRabbitmq {
    ....
}

3，配置ConnectionFactory

    @Bean
    public ConnectionFactory connectionFactory(){

        CachingConnectionFactory factory =  new CachingConnectionFactory("ip",5672);
        factory.setUsername("rabbitmq");
        factory.setPassword("12345678");
        factory.setChannelCacheSize(50);
        return factory;
    }

4，配置amqpAdmin

    @Bean
    public AmqpAdmin amqpAdmin(){

        RabbitAdmin amqpAdmin =  new RabbitAdmin(connectionFactory());
        return amqpAdmin;
    }

5，配置Exchange

    @Bean
    public DirectExchange dex1(){

        return new DirectExchange("dex1");
    }

    @Bean
    public TopicExchange tex1(){

        return new TopicExchange("tex1");
    }

6，配置queue

    @Bean
    public Queue firstQueue(){

        return new Queue("firstQueue");
    }

    @Bean
    public Queue secondQueue(){

        return new Queue("secondQueue");
    }

7，配置binding

    @Bean
    public Binding b1(){

        return BindingBuilder.bind(firstQueue()).to(dex1()).with("key1");
    }
    @Bean
    public Binding b2(){

        return BindingBuilder.bind(secondQueue()).to(tex1()).with("key2");
    }

8，使用declarables维护exchange，queue和bindings，该配置会有amqpAdmin对象进行维护

    @Bean
    public Declarables bs(){

        return new Declarables(new DirectExchange("dex2"),
                new Queue("thirdQueue"),
                new Binding("thirdQueue",DestinationType.QUEUE,"dex2","key3",null));
    }

9，配置RabbitTamplate

    @Bean
    public RabbitTemplate amqpTemplate(){

        RabbitTemplate amqpTemplate = new RabbitTemplate(connectionFactory());

        RetryTemplate retryTemplate = new RetryTemplate();
        ExponentialRandomBackOffPolicy backOffPolicy = new ExponentialRandomBackOffPolicy();
        backOffPolicy.setInitialInterval(500);
        backOffPolicy.setMultiplier(10.0);
        backOffPolicy.setMaxInterval(10000);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        amqpTemplate.setMessageConverter(messageConverter());
        amqpTemplate.setRetryTemplate(retryTemplate);
        return amqpTemplate;
    }

10，配置MessageConverter

    @Bean
    public MessageConverter messageConverter(){

        return new SimpleMessageConverter();
    }

11，监听Queue

    @Bean
    public SimpleMessageListenerContainer container(){

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory());
        //reply 设置
        //  container.setMessageListener(amqpTemplate());
        container.setQueueNames("thirdQueue");
        container.setMessageListener(message -> System.out.println("received message from thirdQueue" +message));
        return container;
    }

12，创建消息发布者

@Component
public class Producer4Amqp{
  
    @Scheduled(fixedDelay = 3000L)
    public void send() {

        System.err.println("sending message"+atomicInteger.get());

        if(atomicInteger.get()%3 ==0){

            this.amqpTemplate.send("dex1","key1",MessageBuilder.withBody((atomicInteger.incrementAndGet()+" message").getBytes()).
                    setContentEncoding(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).setMessageId(atomicInteger.get()+"").build());
        }else if(atomicInteger.get()%3 ==1){

            this.amqpTemplate.send("tex1","key2",MessageBuilder.withBody((atomicInteger.incrementAndGet()+" message").getBytes()).
                    setContentEncoding(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).setMessageId(atomicInteger.get()+"").build());
        }else if(atomicInteger.get()%3 ==2){

            this.amqpTemplate.send("dex2","key3",MessageBuilder.withBody((atomicInteger.incrementAndGet()+" message").getBytes()).
                    setContentEncoding(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).setMessageId(atomicInteger.get()+"").build());
        }


    }
}

13，创建消息消费者

@Component
public class Consumer4Amqp {
  
    @RabbitListener(queues = {"firstQueue"})
    public void received(byte[] message){

        System.out.println("consumer 1 received message from firstQueue" + new String(message));
    }

    @RabbitListener(queues = "secondQueue")
    public void recevied2(Message message){

        System.out.println("consumer 2 received message from secondQueue" + message);
    }
}

启动程序，从console里可以看到如下日志信息：

consumer 1 received message from firstQueue85 message
sending message85
consumer 2 received message from secondQueue(Body:'[B@551d5210(byte[10])' MessageProperties [headers={}, messageId=86, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=tex1, receivedRoutingKey=key2, deliveryTag=39, consumerTag=amq.ctag-cHDXBlrMu0bOgZ4hJSCKbA, consumerQueue=secondQueue])
sending message86
received message from thirdQueue(Body:'[B@585fb560(byte[10])' MessageProperties [headers={}, messageId=87, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=dex2, receivedRoutingKey=key3, deliveryTag=39, consumerTag=amq.ctag-1S_ttl_zj7VdbYHjPScViA, consumerQueue=thirdQueue])
sending message87
consumer 1 received message from firstQueue88 message
sending message88
consumer 2 received message from secondQueue(Body:'[B@52121840(byte[10])' MessageProperties [headers={}, messageId=89, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=tex1, receivedRoutingKey=key2, deliveryTag=40, consumerTag=amq.ctag-cHDXBlrMu0bOgZ4hJSCKbA, consumerQueue=secondQueue])
sending message89
received message from thirdQueue(Body:'[B@451add0(byte[10])' MessageProperties [headers={}, messageId=90, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=dex2, receivedRoutingKey=key3, deliveryTag=40, consumerTag=amq.ctag-1S_ttl_zj7VdbYHjPScViA, consumerQueue=thirdQueue])
sending message90
consumer 1 received message from firstQueue91 message

可以使用rabbitmqctl list_exchanges/list_queus/list_consumers等查看队列状态。