package com.sboot.mq;

import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import java.net.InetAddress;
import java.util.UUID;

public class MQProducerTest extends BaseTest {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    public void test() throws Exception {
        for (int i = 1; i <= 5; i++) {
            MessageProperties messageProperties = new MessageProperties();
            String ip = InetAddress.getLocalHost().getHostAddress();
            messageProperties.setAppId(ip);
//            messageProperties.setUserId(String.valueOf(i));
            MessageConverter messageConverter = new SimpleMessageConverter();
            String msg = UUID.randomUUID().toString();
//            System.out.println(msg);
            Message message1 = messageConverter.toMessage(msg, messageProperties);
            rabbitTemplate.send(MessageQueueConstant.USER_SETTLEMENT_EXCHANGE, "UserSettlementRouting", message1);
            System.out.println("入隊(duì)完成");
            Thread.sleep(500L);
        }
    }
}

package com.sboot.mq;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Component;

import java.net.InetAddress;

@Component
@Profile(value = "dev")
@Slf4j
public class UserSettlementDevConsumer implements ChannelAwareMessageListener {

    @RabbitHandler
    @RabbitListener(queues = MessageQueueConstant.USER_SETTLEMENT_QUEUE, ackMode = "MANUAL")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        Thread.currentThread().setName(UserSettlementDevConsumer.class.getSimpleName() + System.currentTimeMillis());

        long tag = message.getMessageProperties().getDeliveryTag();
        String appId = message.getMessageProperties().getAppId();
        log.info("{}-{}, 消息出隊(duì)", tag, appId);
        String receiveMsg = "";
        try {
            //核對標(biāo)識(shí)，決定是否消費(fèi)消息
            String ip = InetAddress.getLocalHost().getHostAddress();
            if (!ip.equals(appId)) {
                log.info("這不是我需要的消息。放回隊(duì)列。{}", receiveMsg);
//                channel.basicNack(tag, false, true);
                channel.basicReject(tag, true);
//                channel.basicRecover(true);
                return;
            }

            MessageConverter messageConverter = new SimpleMessageConverter();
            receiveMsg = String.valueOf(messageConverter.fromMessage(message));
            。。。。在這里消費(fèi)消息
            log.info("success " + receiveMsg);
            channel.basicAck(tag, false);

        } catch (Exception e) {
            log.error("receive message has an error, ", e);
            channel.basicNack(tag, false, true);
        }
    }

}