SpringBoot之集成RabbitMQ

环境：SpringBoot2.0

<!-- Spring Boot RabbitMQ 依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

yml配置

spring:
    rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        # 开启发送确认
        #publisher-confirms: true
        # 开启发送失败退回
        #publisher-returns: true
        #开启ACK
        #listener.direct.acknowledge-mode: manual
        #listener.simple.acknowledge-mode: manual

RabbitConfig

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
    //topic
    public static final String TOPIC_QUEUE1 = "topic.queue1";
    public static final String TOPIC_QUEUE2 = "topic.queue2";
    public static final String TOPIC_EXCHANGE = "topic.exchange";

    //fanout
    public static final String FANOUT_QUEUE1 = "fanout.queue1";
    public static final String FANOUT_QUEUE2 = "fanout.queue2";
    public static final String FANOUT_EXCHANGE = "fanout.exchange";

    //direct
    public static final String DIRECT_QUEUE1 = "direct.queue1";
    public static final String DIRECT_QUEUE2 ="direct.queue2" ;
    public static final String DIRECT_EXCHANGE = "direct.exchange";


    /**
     * Topic模式
     *
     * @return
     */
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE1);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE2);
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }

    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("zns.message");
    }

    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("zns.#");
    }


    /**
     * Fanout模式
     * Fanout 就是咱们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的全部队列都收到这个消息。
     * @return
     */
    @Bean
    public Queue fanoutQueue1() {
        return new Queue(FANOUT_QUEUE1);
    }

    @Bean
    public Queue fanoutQueue2() {
        return new Queue(FANOUT_QUEUE2);
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE);
    }

    @Bean
    public Binding fanoutBinding1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }

    @Bean
    public Binding fanoutBinding2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

    /**
     * direct模式
     * 消息中的路由键（routing key）若是和 Binding 中的 binding key 一致， 交换器就将消息发到对应的队列中。路由键与队列名彻底匹配
     * @return
     */
    @Bean
    public Queue directQueue1() {
        return new Queue(DIRECT_QUEUE1);
    }

    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(DIRECT_EXCHANGE);
    }

    @Bean
    public Binding directBinding1() {
        return BindingBuilder.bind(directQueue1()).to(directExchange()).with("direct.pwl");
    }

}

 发生消息

import com.zns.admin.api.TestModel;
import com.zns.admin.api.ZNSAdminApiApplication;
import com.zns.admin.api.config.RabbitConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class RabbitmqTest {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    @Test
    public void testTopic() throws Exception {
        TestModel data=new TestModel(1,"topic...");
        this.rabbitTemplate.convertAndSend(RabbitConfig.TOPIC_EXCHANGE,"zns.message", data);
        this.rabbitTemplate.convertAndSend(RabbitConfig.TOPIC_EXCHANGE, "zns.lzc", data);
    }

    @Test
    public void testFanout() throws Exception {
        TestModel data=new TestModel(2,"fanout...");
        this.rabbitTemplate.convertAndSend(RabbitConfig.FANOUT_EXCHANGE, "", data);
    }

    @Test
    public void testDirect() throws Exception {
        TestModel data=new TestModel(3,"direct...");
        this.rabbitTemplate.convertAndSend(RabbitConfig.DIRECT_EXCHANGE, "direct.pwl", data);
    }
}

接收消息

import com.zns.admin.api.config.RabbitConfig;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitmqReceiver {
    @RabbitListener(queues = RabbitConfig.TOPIC_QUEUE1)
    public void receiveTopic1(TestModel data) {
        System.out.println("【receiveTopic1监听到消息】" + data.toString());
    }
    @RabbitListener(queues = RabbitConfig.TOPIC_QUEUE2)
    public void receiveTopic2(TestModel data) {
        System.out.println("【receiveTopic2监听到消息】" + data.toString());
    }

    @RabbitListener(queues = RabbitConfig.FANOUT_QUEUE1)
    public void receiveFanout1(TestModel data) {
        System.out.println("【receiveFanout1监听到消息】" + data.toString());
    }
    @RabbitListener(queues = RabbitConfig.FANOUT_QUEUE2)
    public void receiveFanout2(TestModel data) {
        System.out.println("【receiveFanout2监听到消息】" + data.toString());
    }

    @RabbitListener(queues = RabbitConfig.DIRECT_QUEUE1)
    public void receiveDirect1(TestModel data) {
        System.out.println("【receiveDirect1监听到消息】" + data.toString());
    }
    @RabbitListener(queues = RabbitConfig.DIRECT_QUEUE1)
    public void receiveDirect2(TestModel data) {
        System.out.println("【receiveDirect2监听到消息】" + data.toString());
    }

}

 

ACK消息确认的实现

配置文件打开相关配置开关

spring:
    rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        # 开启发送确认
        publisher-confirms: true
        # 开启发送失败退回
        publisher-returns: true
        #开启ACK
        listener.direct.acknowledge-mode: manual
        listener.simple.acknowledge-mode: manual

这里使用Fanout类型的Exchange为例，主要是设置队列，交换机及绑定

@Configuration
public class RabbitMqFanoutACKConfig {

    @Bean
    public Queue ackQueue() {
        return new Queue("ackQueue");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingAckQueue2Exchange(Queue ackQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(ackQueue).to(fanoutExchange);
    }

}

消息发送服务

@Service
public class AckSenderService implements RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {
        System.out.println("AckSender returnedMessage " + message.toString() + " === " + i + " === " + s1 + " === " + s2);
    }

    /**
     * 消息发送
     */
    public void send() {
        final String content = "如今时间是" + LocalDateTime.now(ZoneId.systemDefault());

        //设置返回回调
        rabbitTemplate.setReturnCallback(this);
        //设置确认回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                System.out.println("消息发送成功！");
            }
            else {
                System.out.println("消息发送失败，" + cause + correlationData.toString());
            }
        });
        rabbitTemplate.convertAndSend("ackQueue", content);
    }
}

消息消费者

@Component
@RabbitListener(queues = {"ackQueue"})
public class MyAckReceiver {

    @RabbitHandler
    public void process(String sendMsg, Channel channel, Message message) {

        System.out.println("AckReceiver  : 收到发送消息 " + sendMsg + ",收到消息时间"
                + LocalDateTime.now(ZoneId.systemDefault()));

        try {
            //告诉服务器收到这条消息已经被当前消费者消费了，能够在队列安全删除，这样后面就不会再重发了，
            //不然消息服务器觉得这条消息没处理掉，后续还会再发
            //第二个参数是消息的标识，false只确认当前一个消息收到，true确认全部consumer得到的消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("process success");
        } catch (Exception e) {
            System.out.println("process fail");
            e.printStackTrace();
        }

    }
}

 调用senderService.send()测试