Amq 使用+springboot

浅浅的花香味﹌ 2023-02-24 08:19 54阅读 0赞

# Amq 使用+springboot #

AMQ是JMS 是java的标准

spring 提供了 JmsTemplate @JmsListerner @EnableJms

## 使用： ##

### 配置 ###

@Configuration
    public class AmqConfig { 
    
    
        //配置Amq的连接池
        @Bean(name = "primaryMqConnectionFactory")
        @Primary
        public ActiveMQConnectionFactory primaryMqConnectionFactory() { 
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
    
            String brokerUrl = environment.getProperty("spring.activemq.broker-url");
            factory.setBrokerURL(brokerUrl);
            String userName = environment.getProperty("spring.activemq.user");
            factory.setUserName(userName);
            String passWord = environment.getProperty("spring.activemq.password");
            factory.setPassword(passWord);
            factory.setCloseTimeout(2000);
            factory.setConnectResponseTimeout(2000);
            factory.setSendTimeout(2000);
           
            return factory;
        }
    
        
        //Amq的jmsTemplate 和jdbc、redis的类似 操作mq
        @Bean(name = "primaryTemplate")
        @Primary
        public JmsTemplate primaryTemplate(@Qualifier("primaryMqConnectionFactory")
                                                       ActiveMQConnectionFactory connectionFactory) { 
            JmsTemplate primaryTemplate = new JmsTemplate(connectionFactory);
            return primaryTemplate;
        }
    
        //topic方式的连接工厂
        @Bean(name = "primaryContainerFactoryTopic")
        public JmsListenerContainerFactory<?> primaryContainerFactoryTopic(@Qualifier("primaryMqConnectionFactory")
                                                                                       ConnectionFactory connectionFactory) { 
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(connectionFactory);
            bean.setPubSubDomain(true);
            return bean;
        }
    
        //queue方式的连接工厂
        @Bean(name = "primaryContainerFactoryQueue")
        public JmsListenerContainerFactory<?> primaryContainerFactoryQueue(@Qualifier("primaryMqConnectionFactory")
                                                                                       ConnectionFactory connectionFactory) { 
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(connectionFactory);
            bean.setPubSubDomain(false);
            return bean;
        }
    }

//MQ的消费者 接口
    public interface NotifyCommonConsumer { 
        
    	 //线程池
         ScheduledExecutorService CONSUMER_POOL = new ScheduledThreadPoolExecutor(2,
                new BasicThreadFactory.Builder().namingPattern("yctm-web-notify-pool-%d").daemon(true).build());
    
        /** * 处理接收消息 * @param connectionFactory * @throws NotifyException */
        void mqttConsumer(NotifyConnectionFactory connectionFactory) throws NotifyException;
    }

//queue消费
    @Component
    public class EventNotifyConsumer extends CommonConsumer { 
       
    	//参数是队列名称 对应的连接工厂
        @JmsListener(destination = ConstantParams.EVENT_QUEUE_NAME, containerFactory = "primaryContainerFactoryQueue")
        public void subscribeAlarmHostEvent(Message message){ 
            //线程池的提交任务处理
            consumerPool.execute(() -> handleRadarMessage(message));
        }
    
        //消息处理函数
        private void handleRadarMessage(Message message) { 
            String event = super.message2String(message);
        }
    }

//topic 消费处理
    @Component
    public class NmsNotifyConsumer extends CommonConsumer { 
    
    // topic的工厂
        @JmsListener(destination = ConstantParams.NMS_NOTIFY_TOPIC, containerFactory = "primaryContainerFactoryTopic")
        public void subscribeAlarmHostEvent(Message message) { 
            consumerPool.execute(() -> handleNmsMessage(message));
        }
    
        private void handleNmsMessage(Message message) { 
        }
    }

生产者

//topic queue 生产者
    @Component
    public class CommonProducer { 
        @Autowired
        private JmsMessagingTemplate primaryTemplate;
    
        public CommonProducer() { 
        }
    
        public void sendMessage(Destination topic, String msg) { 
            this.primaryTemplate.convertAndSend(topic, msg);
        }
    }
    
    //topic 配置
    @Configuration
    public class NmsTopicConfig { 
        @Bean(name = "nmsTopic")
        public Topic nmsTopic(){ 
            return new ActiveMQTopic(ConstantParams.NMS_NOTIFY_TOPIC);
        }
    }
    //queue配置
    @Configuration
    public class RadarQueueConfig { 
        @Bean(name = "radarQueue")
        public Queue radarQueue(){ 
            return new ActiveMQQueue(ConstantParams.EVENT_RADAR_QUEUE_NAME);
        }
    }
    
    @RestController
    @ApiIgnore
    @RequestMapping(value = "/v1/test")
    public class TestAmq { 
        @Resource
        private CommonProducer producer;
        @Autowired
        Topic nmsTopic;
    
        @GetMapping("/test")
        public String getById(@RequestParam Integer id) { 
            //发送消息到topic
            producer.sendMessage(nmsTopic,"aaa");
            return "aaa";
        }
    }