Active MQ 整合SpringBoot

配置文件

POM

<?xml version="1.0" encoding="UTF-8"?>

	4.0.0
	
		org.springframework.boot
		spring-boot-starter-parent
		2.2.3.BUILD-SNAPSHOT
		 
	
	com.ityml.arika
	mq
	0.0.1-SNAPSHOT
	mq
	Demo project for Spring Boot

	
		1.8
	

	
		

		
			org.springframework.boot
			spring-boot-starter-test
			test
			
				
					org.junit.vintage
					junit-vintage-engine
				
			
		
		
			org.springframework.boot
			spring-boot-starter-activemq
		
		
			org.springframework.boot
			spring-boot-starter-web
		
		
		
		
		    org.messaginghub
		    pooled-jms
		
				
		
	            org.apache.commons
	            commons-pool2
	        
		

	
		
			
				org.springframework.boot
				spring-boot-maven-plugin
			
		
	

	
		
			spring-milestones
			Spring Milestones
			https://repo.spring.io/milestone
		
		
			spring-snapshots
			Spring Snapshots
			https://repo.spring.io/snapshot
			
				true
			
		
	
	
		
			spring-milestones
			Spring Milestones
			https://repo.spring.io/milestone
		
		
			spring-snapshots
			Spring Snapshots
			https://repo.spring.io/snapshot
			
				true
			
		
	

yml

server:
  port: 80
  
spring:
  activemq:
    broker-url: tcp://localhost:61616
    user: admin
    password: admin
    
    pool:
      enabled: true
      #连接池最大连接数
      max-connections: 5
      #空闲的连接过期时间，默认为30秒
      idle-timeout: 0
    packages:
      trust-all: true
  jms:
    pub-sub-domain: true

Config类

用于生产ConnectionFactory

package com.ityml.arika;

import javax.jms.ConnectionFactory;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;

@Configuration
@EnableJms
public class ActiveMqConfig {

	 @Bean
	    public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
	        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
	        bean.setPubSubDomain(true);
	        bean.setConnectionFactory(activeMQConnectionFactory);
	        return bean;
	    }
	    // queue模式的ListenerContainer
	    @Bean
	    public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory activeMQConnectionFactory) {
	        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
	        bean.setConnectionFactory(activeMQConnectionFactory);
	        return bean;
	    }
}

收

	@JmsListener(destination = "user",containerFactory = "jmsListenerContainerQueue")
	   public void receiveStringQueue(String msg) {
        System.out.println("接收到消息...." + msg);
    }
	
	@JmsListener(destination = "ooo",containerFactory = "jmsListenerContainerTopic")
	   public void receiveStringTopic(String msg) {
     System.out.println("接收到消息...." + msg);
 }

发

package com.mashibing.arika;

import java.util.ArrayList;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@Service
public class MqProducerService {

	@Autowired
	private JmsMessagingTemplate jmsMessagingTemplate;
	

	
	
	public void sendStringQueue(String destination, String msg) {
		System.out.println("send...");
		ActiveMQQueue queue = new ActiveMQQueue(destination);
		jmsMessagingTemplate.afterPropertiesSet();
		
		ConnectionFactory factory = jmsMessagingTemplate.getConnectionFactory();
		
		try {
			Connection connection = factory.createConnection();
			connection.start();
			
			Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			Queue queue2 = session.createQueue(destination);
			
			MessageProducer producer = session.createProducer(queue2);
			
			TextMessage message = session.createTextMessage("hahaha");
			
			
			producer.send(message);
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		jmsMessagingTemplate.convertAndSend(queue, msg);
	}
	public void sendStringQueueList(String destination, String msg) {
		System.out.println("xxooq");
		ArrayList list = new ArrayList<>();
		list.add("1");
		list.add("2");
		jmsMessagingTemplate.convertAndSend(new ActiveMQQueue(destination), list);
	}
}