ActiveMQ发布订阅模式及Spring整合

水深无声 2022-02-02 04:53 277阅读 0赞

## Topic主题发布和订阅消息 ##

前面讲的案例都是点对点的消息，即一个生产者发送的一条消息只能被一个消费者消费，然后就移除了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70]  
而topic模式一条消息可以被多个消费者订阅，关系如下:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 1]

## 定义生产者 ##

package com.sxt.demo;
    
    import java.net.URI;
    import java.util.Date;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.Topic;
    import javax.jms.TopicConnection;
    import javax.jms.TopicConnectionFactory;
    import javax.jms.TopicPublisher;
    import javax.jms.TopicSession;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQMessage;
    import org.apache.activemq.command.ActiveMQObjectMessage;
    import org.apache.activemq.command.ActiveMQTextMessage;
    
    import com.sxt.bean.User;
    
    /**
     * 定义生产者
     * @author Administrator
     *
     */
    public class ActivemqdemoProducer {
    	private static String userName="admin";
    	private static String password="admin";
    	private static String brokerURL="tcp://192.168.119.12:61616";
    
    	public static void main(String[] args) throws Exception {
    			TopicConnectionFactory factory = new ActiveMQConnectionFactory(userName,password,brokerURL);
    			TopicConnection connection = factory.createTopicConnection();
    			connection.start();
    			
    			TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    			Topic topic = session.createTopic("topic-hello");
    			TopicPublisher publisher = session.createPublisher(topic);
    			 MapMessage message = session.createMapMessage();
    			 message.setString("name", "lisi");
    			 message.setString("age", "18");
    			publisher.send(message );
    			publisher.close();
    			session.close();
    			connection.close();
    
    	}
    }

## 定义消费者 ##

package com.sxt.demo;
    
    import java.net.URI;
    import java.util.Date;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.Topic;
    import javax.jms.TopicConnection;
    import javax.jms.TopicConnectionFactory;
    import javax.jms.TopicPublisher;
    import javax.jms.TopicSession;
    import javax.jms.TopicSubscriber;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQMessage;
    import org.apache.activemq.command.ActiveMQObjectMessage;
    import org.apache.activemq.command.ActiveMQTextMessage;
    
    import com.sxt.bean.User;
    
    /**
     * 定义消费者
     * @author Administrator
     *
     */
    public class ActivemqdemoConsumer {
    	private static String userName="admin";
    	private static String password="admin";
    	private static String brokerURL="tcp://192.168.119.12:61616";
    
    	public static void main(String[] args) throws Exception {
    			TopicConnectionFactory factory = new ActiveMQConnectionFactory(userName,password,brokerURL);
    			TopicConnection connection = factory.createTopicConnection();
    			connection.start();
    			
    			TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    			Topic topic = session.createTopic("topic-hello");
    			TopicPublisher publisher = session.createPublisher(topic);
    			TopicSubscriber subscriber = session.createSubscriber(topic);
    			subscriber.setMessageListener(new MessageListener() {
    				
    				@Override
    				public void onMessage(Message message) {
    					if(message instanceof MapMessage){
    						MapMessage mapMessage=(MapMessage)message;
    						try {
    							String name = mapMessage.getString("name");
    							String age = mapMessage.getString("age");
    							System.out.println(name+"+"+age);
    						} catch (JMSException e) {
    							// TODO Auto-generated catch block
    							System.out.println("错误信息");
    						}
    					}
    					
    				}
    			});
    			Thread.sleep(30000);
    			publisher.close();
    			session.close();
    			connection.close();
    
    	}
    }

## 测试 ##

先启动消费者  
![在这里插入图片描述][20190510170401815.png]  
![在这里插入图片描述][20190510170352167.png]  
启动生产者  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][20190510170533644.png]

## ActiveMQ整合Spring框架 ##

## 整合Spring框架 ##

## 导入jar包 ##

<dependencies>
    	
    	<dependency>
    		<groupId>org.apache.activemq</groupId>
    		<artifactId>activemq-all</artifactId>
    		<version>5.9.0</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.apache.xbean</groupId>
    		<artifactId>xbean-spring</artifactId>
    		<version>4.5</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-jms</artifactId>
    		<version>4.1.6.RELEASE</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-context</artifactId>
    		<version>4.1.6.RELEASE</version>
    	</dependency>
    </dependencies>

## 定义消息载体对象 ##

package com.sxt.bean;
    
    import java.io.Serializable;
    
    public class User implements Serializable{
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = 1L;
    	
    	private String id;
    	private String name;
    	private String password;
    	public String getId() {
    		return id;
    	}
    	public void setId(String id) {
    		this.id = id;
    	}
    	public String getName() {
    		return name;
    	}
    	public void setName(String name) {
    		this.name = name;
    	}
    	public String getPassword() {
    		return password;
    	}
    	public void setPassword(String password) {
    		this.password = password;
    	}
    	@Override
    	public String toString() {
    		return "User [id=" + id + ", name=" + name + ", password=" + password + "]";
    	}
    
    }

## 定义生产者 ##

package com.sxt.producer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    
    import com.sxt.bean.User;
    
    public class OrderProducer {
    		private JmsTemplate template;
    
    		public JmsTemplate getTemplate() {
    			return template;
    		}
    
    		public void setTemplate(JmsTemplate template) {
    			this.template = template;
    		}
    		/**
    		 * 发送消息的方法
    		 */
    		public void sendOrder(String destinationName,User user){
    			template.send(destinationName, new MessageCreator() {
    				/**
    				 * 模板模式中暴露给调用者的方法
    				 */
    				@Override
    				public Message createMessage(Session session) throws JMSException {
    					return session.createObjectMessage(user);
    				}
    			});
    		}
    
    		
    }

## 定义消费者 ##

package com.sxt.consumer;
    
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    
    import org.apache.activemq.command.ActiveMQObjectMessage;
    
    public class OrderConsumer implements MessageListener{
    
    	@Override
    	public void onMessage(Message message) {
    		ActiveMQObjectMessage msg = (ActiveMQObjectMessage) message;
    		try {
    			System.out.println(msg.getObject());
    		} catch (JMSException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		
    	}
    
    }

## Spring配置文件配置 ##

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    	xmlns:amq="http://activemq.apache.org/schema/core"
    	xmlns:jms="http://www.springframework.org/schema/jms"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans     
            http://www.springframework.org/schema/beans/spring-beans-4.0.xsd     
            http://www.springframework.org/schema/context     
            http://www.springframework.org/schema/context/spring-context-4.0.xsd  
            http://www.springframework.org/schema/jms  
            http://www.springframework.org/schema/jms/spring-jms-4.0.xsd  
            http://activemq.apache.org/schema/core  
            http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd">
    	
    	
    	
    	<amq:connectionFactory id="amqConnectionFactory"
    		brokerURL="tcp://192.168.119.12:61616" userName="admin" password="admin" />
    
    	
    	
    	<bean id="connectionFactory"
    		class="org.springframework.jms.connection.CachingConnectionFactory">
    		
    		<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
    		
    		<property name="sessionCacheSize" value="100" />
    	</bean>
    
    	
    
    	
    	<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
    		
    		<constructor-arg ref="connectionFactory" />
    		
    		
    	</bean>
    	
    	
    	<bean id="orderProducer" class="com.sxt.producer.OrderProducer">
    		
    		<property name="template" ref="jmsQueueTemplate"></property>
    	</bean>
    	
    	
    
    	
    
    	
    	<jms:listener-container destination-type="queue"
    		container-type="default" connection-factory="connectionFactory"
    		acknowledge="auto">
    		
    		<jms:listener destination="spring-MQ" ref="orderReciver" />
    	</jms:listener-container>
    	
    	
    	<bean id="orderReciver" class="com.sxt.consumer.OrderConsumer"/>
    
    	
    </beans>

## 测试 ##

package com.sxt.test;
    
    import org.apache.xbean.spring.context.ClassPathXmlApplicationContext;
    import org.springframework.context.ApplicationContext;
    
    import com.sxt.bean.User;
    import com.sxt.producer.OrderProducer;
    
    public class Test {
    	public static void main(String[] args) {
    		ApplicationContext ac = new ClassPathXmlApplicationContext("application.xml");
    		OrderProducer bean = ac.getBean(OrderProducer.class);
    		User user = new User();
    		user.setId("2");
    		user.setName("lisi");
    		user.setPassword("123456");
    		bean.sendOrder("spring-MQ", user);
    		System.out.println("消息发送完成");
    		
    	}
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][20190510170955744.png]

## 整合完成 ##

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70]: /images/20220202/4a14a17e25ba40a8a3bbea93f20b40e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 1]: /images/20220202/d2897875a0e2480e9a697b8ea33c4d08.png
[20190510170401815.png]: /images/20220202/1444d85504b343f5906ddb5f0f2f1a41.png
[20190510170352167.png]: /images/20220202/4592c9033e084b8d9e66e080a8d65f54.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 2]: /images/20220202/7729320119824bffb98b94c6f58a7f05.png
[20190510170533644.png]: /images/20220202/19b9e6faa008486ea512ccc3d5449901.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 3]: /images/20220202/689626678fc64638af7528a5266d5ff6.png
[20190510170955744.png]: /images/20220202/d6c877d342cd4ae9875547a93007d679.png