SpringBoot 注入RabbitMQ java初级使用

package com.cetcnav.config.rabbitmq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Description:
 * @Author: dell
 * @Date: 2024/5/8 13:21
 */
@Configuration
public class DirectRabbitConfig {

    //队列名：TestDirectQueue
    @Bean
    public Queue QXBDWarningMsgQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        //   return new Queue("TestDirectQueue",true,true,false);

        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("QXBD_WARNING_MSG_QUEUE", true);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange QXBDWarningMsgExchange() {
        return new DirectExchange("QXBD_WARNING_MSG_QUEUE_EXCHANGE", true, false);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(QXBDWarningMsgQueue()).to(QXBDWarningMsgExchange()).with("QXBD_WARNING_MSG_QUEUE_ROUTING");
    }
}

package com.cetcnav.rabbitmqListener;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cetcnav.cache.SkyMirrorDataCache;
import com.cetcnav.domain.DI.DIData;
import com.cetcnav.domain.skyMirrorData.SkyMirrorData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.ParseException;

/**
 * @Description:
 * @Author: dell
 * @Date: 2024/5/8 9:24
 */
@Component
public class DirectReceiver {

    private static Logger log = LoggerFactory.getLogger(DirectReceiver.class);

    // 单条发送
    @Value("${skymirror.pushUrl}")
    private String PUSH_URL;
    // 批量发送
    @Value("${skymirror.pushBatchUrl}")
    private String PUSH_BATCH_URL;

    @RabbitListener(queues = "QXBD_WARNING_MSG_QUEUE")
    public void receiveMessage(String message) throws ParseException {
        log.info("queue.consumer 收到1条message:{}", message);
        try {
            // 处理RabbitMQ接收的状态告警信息
            dealMessage(message);
        } catch (Exception e) {
            log.error(ExceptionUtil.getMessage(e), e);
        }
    }


    /**
     * 处理RabbitMQ接收的状态告警信息
     */
    private void dealMessage(String message) {
        SkyMirrorData skyMirrorData = JSON.parseObject(message, SkyMirrorData.class);
        SkyMirrorDataCache.addSkyMirrorDataCache(skyMirrorData);
    }
}

package com.cetcnav.rabbitmqService;

import com.alibaba.fastjson.JSONObject;
import com.cetcnav.domain.skyMirrorData.SkyMirrorData;
import com.cetcnav.domain.skyMirrorData.SkyMirrorData_old;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @Description:
 * @Author: dell
 * @Date: 2024/5/8 13:52
 */
@Component
public class RabbitmqService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 推送消息
    public boolean sendDirectMsg(SkyMirrorData skyMirrorData) {
        rabbitTemplate.convertAndSend("QXBD_WARNING_MSG_QUEUE_EXCHANGE", "QXBD_WARNING_MSG_QUEUE_ROUTING",
                JSONObject.toJSONString(skyMirrorData));
        return true;
    }

}

原文地址：https://blog.csdn.net/qq_33651286/article/details/140640568

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！