RabbitMQ 简单模式

配置RabbitMQ

@Configuration
public class RabbitMQConfig {

    @Bean
    public ConnectionFactory getFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setVirtualHost("/zwe-api");
        return factory;
    }
}

生产者:

@Log
@Component
public class SimpleMessageTask {


    @Resource
    ConnectionFactory faction;

    @Resource
    PersonalMessageDao personalMessageDao;


    public void send(String topic) {
        try (Connection connection = faction.newConnection();
             Channel channel = connection.createChannel();
             ) {
            channel.queueDeclare(topic, true, false, false, null);
            String msg = "Hello Ja7";
            // 此处可设置设置消息头
            HashMap hashMap = new HashMap();
            hashMap.put("id", "10000");
            AMQP.BasicProperties build = new AMQP.BasicProperties().builder().headers(hashMap).build();
            channel.basicPublish("", topic, build, msg.getBytes("UTF-8"));
        } catch (Exception e) {
            e.printStackTrace();
            throw new ZweApiException("消息队列发布失败");
        }
    }

消费之

@Log
@Component
public class SimpleMessageTask {

    @Resource
    ConnectionFactory faction;

    public void receive(String topic) {
        int i = 0; // ack应答后才会执行下一个
        try (Connection connection = faction.newConnection();
             Channel channel = connection.createChannel()
        ) {
            channel.queueDeclare(topic, true, false, false, null);
            // 循环将队列里的消息全部读完
            while (true) { 
                GetResponse getResponse = channel.basicGet(topic, false);
                if (getResponse != null) {
                    AMQP.BasicProperties props = getResponse.getProps();
                    Map headers = props.getHeaders();
                    // 拿到头信息
                    String id = headers.get("id").toString();

                    // 消息体
                    byte[] body = getResponse.getBody();
                    String msg = new String(body, "UTF-8");
                    
                    // 业务代码

                    
                    // ack应答
                    long deliveryTag = getResponse.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);

                    i++;
                } else {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new ZweApiException("接收消息失败");
        }
    }