RabbitMQ 快速入門

時間 2019-12-06

原文原文鏈接

前面咱們介紹了RabbitMQ的基本概念，RabbitMQ基礎概念詳細介紹。在這裏咱們作一個簡單的例子進行快速入門。java

新建Spring Boot項目引入依賴包

<parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>1.5.3.RELEASE</version>
	<relativePath /> <!-- lookup parent from repository -->
</parent>

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

在啓動類上添加啓動MQ的註解

@EnableRabbit

添加配置

# Rabbitmq
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=test
spring.rabbitmq.addresses=192.168.35.128:5672
#spring.rabbitmq.addresses=192.168.35.128:5672,192.168.35.129:5672,192.168.35.130:5672
spring.rabbitmq.connection-timeout=50000
#rabbitmq listetner
# 消費者最小數量
spring.rabbitmq.listener.concurrency=10
# 消費者最大數量
spring.rabbitmq.listener.max-concurrency=20
# 消息的確認模式
spring.rabbitmq.listener.acknowledge-mode=MANUAL
# 每一次發送到消費者的消息數量，它應該大於或等於事務大小（若是使用）。
spring.rabbitmq.listener.prefetch=10
# 消費者端的重試
spring.rabbitmq.listener.retry.enabled=true
#rabbitmq publisher
# 生產者端的重試
spring.rabbitmq.template.retry.enabled=true
#開啓發送消息到exchange確認機制
spring.rabbitmq.publisher-confirms=true
#開啓發送消息到exchange可是exchange沒有和隊列綁定的確認機制
spring.rabbitmq.publisher-returns=true

RabbitMQ 全部配置參考：git

# RABBIT (RabbitProperties)
spring.rabbitmq.addresses= # Comma-separated list of addresses to which the client should connect.
spring.rabbitmq.cache.channel.checkout-timeout= # Number of milliseconds to wait to obtain a channel if the cache size has been reached.
spring.rabbitmq.cache.channel.size= # Number of channels to retain in the cache.
spring.rabbitmq.cache.connection.mode=CHANNEL # Connection factory cache mode.
spring.rabbitmq.cache.connection.size= # Number of connections to cache.
spring.rabbitmq.connection-timeout= # Connection timeout, in milliseconds; zero for infinite.
spring.rabbitmq.dynamic=true # Create an AmqpAdmin bean.
spring.rabbitmq.host=localhost # RabbitMQ host.
spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container.
spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup.
spring.rabbitmq.listener.concurrency= # Minimum number of consumers.
spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`.
spring.rabbitmq.listener.max-concurrency= # Maximum number of consumers.
spring.rabbitmq.listener.prefetch= # Number of messages to be handled in a single request. It should be greater than or equal to the transaction size (if used).
spring.rabbitmq.listener.retry.enabled=false # Whether or not publishing retries are enabled.
spring.rabbitmq.listener.retry.initial-interval=1000 # Interval between the first and second attempt to deliver a message.
spring.rabbitmq.listener.retry.max-attempts=3 # Maximum number of attempts to deliver a message.
spring.rabbitmq.listener.retry.max-interval=10000 # Maximum interval between attempts.
spring.rabbitmq.listener.retry.multiplier=1.0 # A multiplier to apply to the previous delivery retry interval.
spring.rabbitmq.listener.retry.stateless=true # Whether or not retry is stateless or stateful.
spring.rabbitmq.listener.transaction-size= # Number of messages to be processed in a transaction. For best results it should be less than or equal to the prefetch count.
spring.rabbitmq.password= # Login to authenticate against the broker.
spring.rabbitmq.port=5672 # RabbitMQ port.
spring.rabbitmq.publisher-confirms=false # Enable publisher confirms.
spring.rabbitmq.publisher-returns=false # Enable publisher returns.
spring.rabbitmq.requested-heartbeat= # Requested heartbeat timeout, in seconds; zero for none.
spring.rabbitmq.ssl.enabled=false # Enable SSL support.
spring.rabbitmq.ssl.key-store= # Path to the key store that holds the SSL certificate.
spring.rabbitmq.ssl.key-store-password= # Password used to access the key store.
spring.rabbitmq.ssl.trust-store= # Trust store that holds SSL certificates.
spring.rabbitmq.ssl.trust-store-password= # Password used to access the trust store.
spring.rabbitmq.ssl.algorithm= # SSL algorithm to use. By default configure by the rabbit client library.
spring.rabbitmq.template.mandatory=false # Enable mandatory messages.
spring.rabbitmq.template.receive-timeout=0 # Timeout for `receive()` methods.
spring.rabbitmq.template.reply-timeout=5000 # Timeout for `sendAndReceive()` methods.
spring.rabbitmq.template.retry.enabled=false # Set to true to enable retries in the `RabbitTemplate`.
spring.rabbitmq.template.retry.initial-interval=1000 # Interval between the first and second attempt to publish a message.
spring.rabbitmq.template.retry.max-attempts=3 # Maximum number of attempts to publish a message.
spring.rabbitmq.template.retry.max-interval=10000 # Maximum number of attempts to publish a message.
spring.rabbitmq.template.retry.multiplier=1.0 # A multiplier to apply to the previous publishing retry interval.
spring.rabbitmq.username= # Login user to authenticate to the broker.
spring.rabbitmq.virtual-host= # Virtual host to use when connecting to the broker.

聲明隊列

@Configuration
@ConditionalOnBean({RabbitTemplate.class})
public class RabbitConfig {

    /**
     * 方法rabbitAdmin的功能描述:動態聲明queue、exchange、routing
     *
     * @param connectionFactory
     * @return
     * @author : yuhao.wang
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 發放獎勵隊列交換機
        DirectExchange exchange = new DirectExchange(RabbitConstants.MQ_EXCHANGE_SEND_AWARD);

        //聲明發送優惠券的消息隊列（Direct類型的exchange）
        Queue couponQueue = queue(RabbitConstants.QUEUE_NAME_SEND_COUPON);
        rabbitAdmin.declareQueue(couponQueue);
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(BindingBuilder.bind(couponQueue).to(exchange).with(RabbitConstants.MQ_ROUTING_KEY_SEND_COUPON));

        return rabbitAdmin;
    }

    public Queue queue(String name) {
        // 是否持久化
        boolean durable = true;
        // 僅建立者可使用的私有隊列，斷開後自動刪除
        boolean exclusive = false;
        // 當全部消費客戶端鏈接斷開後，是否自動刪除隊列
        boolean autoDelete = false;
        return new Queue(name, durable, exclusive, autoDelete, args);
    }
}

在這裏咱們申明瞭一個RabbitConstants.QUEUE_NAME_SEND_COUPON隊列，並聲明瞭一個DirectExchange 類型的交換器，經過Bind將隊列、交換機和路由RabbitConstants.MQ_ROUTING_KEY_SEND_COUPON的關係進行綁定。github

消息的生產者

/**
 * Rabbit 發送消息
 *
 * @author yuhao.wang
 */
@Service
public class RabbitSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback, InitializingBean {
    private final Logger logger = LoggerFactory.getLogger(RabbitSender.class);

    /**
     * Rabbit MQ 客戶端
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 系統配置
     */
    @Autowired
    private SystemConfig systemConfig;

    /**
     * 發送MQ消息
     *
     * @param exchangeName 交換機名稱
     * @param routingKey   路由名稱
     * @param message      發送消息體
     */
    public void sendMessage(String exchangeName, String routingKey, Object message) {

        // 獲取CorrelationData對象
        CorrelationData correlationData = this.correlationData(message);
        rabbitTemplate.convertAndSend(exchangeName, routingKey, message, correlationData);
    }

    /**
     * 用於實現消息發送到RabbitMQ交換器後接收ack回調。
     * 若是消息發送確認失敗就進行重試。
     *
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(org.springframework.amqp.rabbit.support.CorrelationData correlationData, boolean ack, String cause) {
        // 消息回調確認失敗處理
        if (!ack) {
            // 這裏以作消息的從發等處理
            logger.info("消息發送失敗，消息ID:{}", correlationData.getId());
        } else {
            logger.info("消息發送成功,消息ID:{}", correlationData.getId());
        }
    }

    /**
     * 用於實現消息發送到RabbitMQ交換器，但無相應隊列與交換器綁定時的回調。
     * 基本上來講線程不可能出現這種狀況，除非手動將已經存在的隊列刪掉，不然在測試階段確定能測試出來。
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.error("MQ消息發送失敗，replyCode:{}, replyText:{}，exchange:{}，routingKey:{}，消息體:{}",
                replyCode, replyText, exchange, routingKey, JSON.toJSONString(message.getBody()));

        // TODO 保存消息到數據庫
    }

    /**
     * 消息相關數據（消息ID）
     *
     * @param message
     * @return
     */
    private CorrelationData correlationData(Object message) {

        return new CorrelationData(UUID.randomUUID().toString(), message);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }
}

ConfirmCallback和ReturnCallback

在這個裏咱們主要實現了ConfirmCallback和ReturnCallback兩個接口。這兩個接口主要是用來發送消息後回調的。由於rabbit發送消息是隻管發，至於發沒發成功，發送方法無論。spring

ConfirmCallback：當消息成功到達exchange的時候觸發的ack回調。
ReturnCallback：當消息成功到達exchange，可是沒有隊列與之綁定的時候觸發的ack回調。基本上來講線上不可能出現這種狀況，除非手動將已經存在的隊列刪掉，不然在測試階段確定能測試出來。

若是使用RabbitMQ的ConfirmCallback和ReturnCallback模式必須將下面兩個開關打開，不然將不生效：數據庫

# 生產者端的重試
spring.rabbitmq.template.retry.enabled=true
#開啓發送消息到exchange確認機制
spring.rabbitmq.publisher-confirms=true

消息的發送方式

rabbitTemplate.send(message); //發消息，參數類型爲org.springframework.amqp.core.Message
rabbitTemplate.convertAndSend(object); //轉換併發送消息。將參數對象轉換爲org.springframework.amqp.core.Message後發送，這個是異步的。消息是否發送成功須要用到ConfirmCallback和ReturnCallback回調函數類確認。
rabbitTemplate.convertSendAndReceive(message) //轉換併發送消息,且等待消息者返回響應消息。這是一個RPC方法，當發送消息事後，該方法會一直阻塞在哪裏等待返回結果，直到請求超時。能夠經過配置spring.rabbitmq.template.reply-timeout來配置超時時間。

消息的消費者

/**
 * 發放優惠券的MQ處理
 *
 * @author yuhao.wang
 */
public class SendMessageListener {

    private final Logger logger = LoggerFactory.getLogger(SendMessageListener.class);

    @RabbitListener(queues = RabbitConstants.QUEUE_NAME_SEND_COUPON)
    public void process(SendMessage sendMessage, Channel channel, Message message) throws Exception {
        try {
            // 參數校驗
            Assert.notNull(sendMessage, "sendMessage 消息體不能爲NULL");

             logger.info("處理MQ消息");
             // 確認消息已經消費成功
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            logger.error("MQ消息處理異常，消息ID：{}，消息體:{}", message.getMessageProperties().getCorrelationIdString(),
                    JSON.toJSONString(sendMessage), e);

            // 拒絕當前消息，並把消息返回原隊列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        } 
    }
}

使用 @RabbitListener註解，並在註解上指定你要監聽的隊列名稱，這樣子消費者就聲明好了。這裏有兩點要注意一下：併發

監聽消息的參數（如上面的sendMessage參數）必定要和發送的時候相匹配，也能夠不使用sendMessage參數，直接在Message參數裏面獲取消息體。
若是有返回信息，直接return就好（rabbitTemplate.convertSendAndReceive()方法就會有返回值）。

消息的確認和拒絕

消息確認

我看能夠調用Channel類中的basicAck方法進行消息確認，方法定義以下：app

void basicAck(long deliveryTag, boolean multiple) throws IOException;

deliveryTag：消息編號。
multiple：是否確認多條消息。false，確認當前這條消息；true，確認deliveryTag編號之前的全部消息。

消息拒絕

拒絕消息可使用Channel中的basicReject或者basicNack方法，basicReject只能拒絕一條消息，basicNack能夠拒絕多條消息。less

void basicReject(long deliveryTag, boolean requeue) throws IOException;

void basicNack(long deliveryTag, boolean multiple, boolean requeue)
            throws IOException;