訊息生產者和消費者

import com.rabbitmq.client.Channel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.io.IOException;
import java.util.Date;
import java.util.UUID;

/**
 * Created by yangliu on 2018/4/8.
 */
@Controller
@RequestMapping("/rabbitMq")
public class TestController {
    private Logger logger= LoggerFactory.getLogger(TestController.class);
    @Autowired
    RabbitAdmin rabbitAdmin;

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.getRabbitTemplate().setConfirmCallback(new MsgSendConfirmCallBack());
        rabbitAdmin.getRabbitTemplate().setReturnCallback(new MsgSendReturnCallback());
        return rabbitAdmin;
    }


    @RequestMapping("/sendMq")
    @ResponseBody
    public String send(String name) throws Exception {
        String context = "hello "+name+" --" + new Date();
        String sendStr;
        for(int i=1;i<=100;i++){
            sendStr="第["+i+"]個 hello  --" + new Date();
            logger.debug("HelloSender: " + sendStr);
            sendMessage("myqueue",sendStr);
            //Thread.sleep(1000);
        }
        return context;
    }

    /**
     * 方式一：動態宣告exchange和queue它們的繫結關係  rabbitAdmin
     * @param exchangeName
     * @param queueName
     */
    protected void declareBinding(String exchangeName, String queueName) {
        if (rabbitAdmin.getQueueProperties(queueName) == null) {
            /*  queue 佇列宣告
            durable=true,交換機持久化,rabbitmq服務重啟交換機依然存在,保證不丟失; durable=false,相反
            auto-delete=true:無消費者時，佇列自動刪除; auto-delete=false：無消費者時，佇列不會自動刪除
            排他性，exclusive=true:首次申明的connection連線下可見; exclusive=false：所有connection連線下*/
            Queue queue = new Queue(queueName, true, false, false, null);
            rabbitAdmin.declareQueue(queue);
            TopicExchange directExchange = new TopicExchange(exchangeName);
            rabbitAdmin.declareExchange(directExchange);//宣告exchange
            Binding binding = BindingBuilder.bind(queue).to(directExchange).with(queueName);    //將queue繫結到exchange
            rabbitAdmin.declareBinding(binding);      //宣告繫結關係
            rabbitAdmin.getRabbitTemplate().setMandatory(true);
            rabbitAdmin.getRabbitTemplate().setConfirmCallback(new MsgSendConfirmCallBack());//訊息確認
            rabbitAdmin.getRabbitTemplate().setReturnCallback(new MsgSendReturnCallback());//確認後回撥
        } else {
            rabbitAdmin.getRabbitTemplate().setQueue(queueName);
            rabbitAdmin.getRabbitTemplate().setExchange(queueName);
            rabbitAdmin.getRabbitTemplate().setRoutingKey(queueName);
        }
    }

    /**
     * 傳送訊息
     * @param queueName
     * @param message
     * @throws Exception
     */
    public void sendMessage(String queueName, String message) throws Exception {
        declareBinding(queueName, queueName);
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        rabbitAdmin.getRabbitTemplate().convertAndSend(queueName, queueName, message,correlationId);
        logger.debug("[rabbitmq-sendMessage]queueName:{} ,uuid:{},msg:{}",queueName,correlationId.getId(),message);
    }

    /**
     * 消費者
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        Queue queue = new Queue("myqueue", true, false, false, null);
        container.setQueues(queue);
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //設定確認模式手工確認
        container.setMessageListener(new ChannelAwareMessageListener() {
        @Override
        public void onMessage(Message message, Channel channel) throws Exception {

                    byte[] body = message.getBody();
                    try {
                        //業務邏輯
                        logger.info("消費 receive msg : " + new String(body));
                        // 訊息的標識，false只確認當前一個訊息收到，true確認所有consumer獲得的訊息
                        //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //手動確認確認訊息成功消費
                    } catch (Exception e) {
                        logger.info("消費失敗: " + new String(body));
                        // ack返回false，並重新回到佇列，api裡面解釋得很清楚
                        try {
                            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                        } catch (IOException e1) {
                            e1.printStackTrace();
                        }
                    }

        }
        });
        return container;
    }


    /*

    //訊息的標識，false只確認當前一個訊息收到，true確認所有consumer獲得的訊息
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    //ack返回false，並重新回到佇列，api裡面解釋得很清楚
    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    //拒絕訊息
    channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);

    如果訊息沒有到exchange,則confirm回撥,ack=false
    如果訊息到達exchange,則confirm回撥,ack=true
    exchange到queue成功,則不回撥return
    exchange到queue失敗,則回撥return(需設定mandatory=true,否則不回回調,訊息就丟了)
    */
}
 

失敗後return回撥:

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
public class MsgSendReturnCallback implements RabbitTemplate.ReturnCallback {
    @Override  
    public void returnedMessage(Message message, int replyCode, String replyText,
                                String exchange, String routingKey) {
        System.out.println("確認後回撥return--message:"+new String(message.getBody())+",replyCode:"+replyCode+",replyText:"
                +replyText+",exchange:"+exchange+",routingKey:"+routingKey);
    }
}  
 

確認後回撥: 

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;

public class MsgSendConfirmCallBack implements RabbitTemplate.ConfirmCallback {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if (ack) {
                System.out.println("訊息確認成功cause"+cause);
            } else {
                //處理丟失的訊息
                System.out.println("訊息確認失敗:"+correlationData.getId()+"#cause"+cause);
            }
        }
    }