springboot 整合activeMq

消息队列概述

常用的消息队列的对比

整合的步骤

业务场景，用户登录之后 给用户 发送短信，邮件，增加积分

pom文件配置

	   <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>

配置文件 增加mq服务端信息

	spring.activemq.broker-url=tcp://127.0.0.1:61616

启动类上需要开启消息队列模式

@EnableJms //开启消息队列模式
public class SpringbootUserRegisterApplication {

	public static void main(String[] args) {
		SpringApplication.run(SpringbootUserRegisterApplication.class, args);
	}

}

消息生产者代码

注入JmsTemplate

		@Autowired
 JmsTemplate jmsTemplate;

增加消息生产者的需求

	@RequestMapping("login")
 public String register(String uname) {
 	System.out.println("登录校验，校验用户名 密码");
 	
 	//去发短信
 	Destination message = new ActiveMQQueue("message.queen");
 	jmsTemplate.convertAndSend(message, uname);

 	//去发邮件
 	Destination email = new ActiveMQTopic("email.topic");
 	jmsTemplate.convertAndSend(email, uname);
 	
 	//去增加积分
 	Destination score_point = new ActiveMQQueue("score.queen");
 	jmsTemplate.convertAndSend(score_point, uname);
 	return "redirect:list";
 }

备注

ActiveMQQueue 为队列模式消息
ActiveMQTopic 为订阅模式的消息

消息消费者 即使接受者案例

pom文件中 需要增加 与消息发送者 同样的依赖

消息模式默认是队列(queen)模式

开始订阅(topic)模式

#topic 模式
#spring.jms.pub-sub-domain=true

如何需要项目同事支持 queen 和topic模式 ，需要增加配置类

@Configuration
public class ActiveConfig {
	 /**
     * 处理topic消息
     * @param connectionFactory
     * @return
     */
    @Bean
    public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(true);
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }

    /**
     * 处理queue消息
     * @param connectionFactory
     * @return
     */
    @Bean
    public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(false);
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
}

消息的业务处理代码

@JmsListener(destination="message.queen",containerFactory = "queueListenerFactory")
	public void messageSend(String text) throws UnknownHostException {
		InetAddress   address = InetAddress.getLocalHost();
		System.out.println("端口号"+address.getHostAddress()+",ip:"+ip.getPort());
	    System.out.println("接收到消息类型为queen：为" + text+"发送短信");
	}
	
	
	@JmsListener(destination="email.topic", containerFactory = "topicListenerFactory")
	public void emailSend(String text) throws UnknownHostException {
		InetAddress   address = InetAddress.getLocalHost();
		System.out.println("端口号"+address.getHostAddress()+",ip:"+ip.getPort());
	    System.out.println("接收到消息类型为topic：为" + text+"发送邮件");
	}
	
	@JmsListener(destination="score.queen",containerFactory = "queueListenerFactory")
	public void scoreAdd(String text) throws UnknownHostException {
		InetAddress   address = InetAddress.getLocalHost();
		System.out.println("端口号"+address.getHostAddress()+",ip:"+ip.getPort());
		System.out.println("接收到消息：为" + text+"增加积分");
	}

测试步骤

运行登录功能

运行消息接受者