2.9、ActiveMQ ——消息选择器 Selector

水深无声 2021-12-10 19:57 340阅读 0赞

### 文章目录 ###

*   *   *  前言
         *  ActiveMQ 中的 Selecor
         *   *  示意图
             *  Topic 和 Queue 的区别
         *  代码
         *   *  Queue 类型中使用 Selector
         *  Queue 类型的 生产者
         *  Queue 类型的 消费者
         *   *  Topic 类型中使用 Selector
         *  Topic 类型的生产者
         *  Topic 类型的消费者
         *  总结

### 前言 ###

> 体能状态先于精神状态，习惯先于决心，聚焦先于喜好。

### ActiveMQ 中的 Selecor ###

> ActiveMQ 允许使用客户端通过消息选择器对消息进行过滤处理。  
> Queue 和 Topic 都支持，但又有微妙的区别。

#### 示意图 ####

> 首先在生产者端，需要为每一条发送的消息增加一个属性 键值对，比如下图的 name=‘A’ 和 name=‘B’  
> 其次在客户端需要在创建消费者时指定选择器条件——name='A’或name=‘B’  
> 由于Topic和Queue的区别，当两个不同选择条件的客户端对同一个Topic或Queue进行消息获取时，Topic之间不会有什么影响，但是对于Queue，由于此时它们共同处理同一个队列中的消息，一旦某一个消费者出现了故障，会造成另一个消费者处理速度变慢，因为出现故障的消费者的消息一直在队列的前端，阻塞的消息可能越来越多。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70]

#### Topic 和 Queue 的区别 ####

> 两个消费者消费同一个 Queue ，选择器条件不同，Queue中每一条消息只能被其中一个消费者消费（不考虑消息确认机制）  
> Topic 类型消息，两个消费者订阅同一个Topic ,两个消费者之间不会有什么影响。

### 代码 ###

#### Queue 类型中使用 Selector ####

### Queue 类型的 生产者 ###

> 需要给发送的消息增加一个属性 message.setStringProperty(“name”, “A”);  
> 这样消费者才可以用以做消息选择器的条件

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    /** * 生产者生产特定消息——消费者可以通过消息选择器进行选择 * @author jie.wu */
    public class ActiveMQSelectorQueueSendTest { 
    
    	@Test
    	public void send() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		try { 
    			//创建连接
    			connection=factory.createConnection();
    			//开启连接
    			connection.start();
    			//创建 session
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//创建 Queue
    			Queue queue=session.createQueue("queue001");
    			//创建生产者
    			MessageProducer producer=session.createProducer(queue);
    			
    			//第一条消息
    			TextMessage message=session.createTextMessage("this is a message");
    			message.setStringProperty("name", "A");
    			System.out.println("发送消息A:"+System.currentTimeMillis());
    			producer.send(message);
    			
    			//第二条消息
    			TextMessage message2=session.createTextMessage("this is a message");
    			message.setStringProperty("name", "B");
    			System.out.println("发送消息B:"+System.currentTimeMillis());
    			producer.send(message);
    			
    		}  catch (JMSException e) { 
    			System.out.println("出现问题:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    				if(session!=null) { 
    					session.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭连接出现问题:"+e);
    			}
    		}
    		
    	}
    }

### Queue 类型的 消费者 ###

> 需要在创建消费者时指定选择条件  
> MessageConsumer consumer1=session.createConsumer(queue, “name=‘B’”);  
> 可以从运行结果看到，消息选择器可以改变Queue 中消息默认的处理顺序，即你可以直接选择你想要的消息，略过你不想要的消息。

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    /** * 消费者消费通队列特定消息——通过消息选择器 * @author jie.wu */
    public class ActiveMQSelectorQueueReceiveTest { 
    	
    	@Test
    	public void receive() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//创建连接
    			connection=factory.createConnection();
    			//开启连接
    			connection.start();
    			//创建 session
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//创建 Queue
    			Queue queue=session.createQueue("queue001");
    			//创建消费者1-增加 selector name='B'
    			MessageConsumer consumer1=session.createConsumer(queue, "name='B'");
    			TextMessage message1=(TextMessage)consumer1.receive();
    			System.out.println("先获取 name='B':"+message1.getText());
    			
    			//创建消费者2- 增加 selector name='A'
    			MessageConsumer consumer2=session.createConsumer(queue, "name='A'");
    			TextMessage message2=(TextMessage)consumer2.receive();
    			System.out.println("再获取 name='A':"+message2.getText());
    			
    		}  catch (JMSException e) { 
    			System.out.println("出现问题:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    				if(session!=null) { 
    					session.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭连接出现问题:"+e);
    			}
    		}
    		
    	}
    
    }

#### Topic 类型中使用 Selector ####

### Topic 类型的生产者 ###

> 需要给发送的消息增加一个属性 message.setStringProperty(“name”, “A”);  
> 这样消费者才可以用以做消息选择器的条件

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQSelecorTopicSendTest { 
    	@Test
    	public void send() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		try { 
    			//创建连接
    			connection=factory.createConnection();
    			//开启连接
    			connection.start();
    			//创建 session
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//创建 Topic
    			Topic topic=session.createTopic("topic001");
    			//创建生产者
    			MessageProducer producer=session.createProducer(topic);
    			
    			//第一条消息
    			TextMessage message=session.createTextMessage("this is a message");
    			message.setStringProperty("name", "A");
    			System.out.println("发送消息A:"+System.currentTimeMillis());
    			producer.send(message);
    			
    			//第二条消息
    			TextMessage message2=session.createTextMessage("this is a message");
    			message.setStringProperty("name", "B");
    			System.out.println("发送消息B:"+System.currentTimeMillis());
    			producer.send(message);
    			
    		}  catch (JMSException e) { 
    			System.out.println("出现问题:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    				if(session!=null) { 
    					session.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭连接出现问题:"+e);
    			}
    		}
    		
    	}
    
    }

### Topic 类型的消费者 ###

> 需要在创建消费者时指定选择条件  
> MessageConsumer consumer1=session.createConsumer(topic, “name=‘B’”);

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQSelecorTopicReceiveTest { 
    	@Test
    	public void receive1() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//创建连接
    			connection=factory.createConnection();
    			//开启连接
    			connection.start();
    			//创建 session
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//创建 Topic
    			Topic topic=session.createTopic("topic001");
    			//创建消费者1-增加 selector name='B'
    			MessageConsumer consumer1=session.createConsumer(topic, "name='B'");
    			TextMessage message1=(TextMessage)consumer1.receive();
    			System.out.println("topic获取 name='B':"+message1.getText());
    			
    		}  catch (JMSException e) { 
    			System.out.println("出现问题:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    				if(session!=null) { 
    					session.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭连接出现问题:"+e);
    			}
    		}
    		
    	}
    	
    	@Test
    	public void receive2() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//创建连接
    			connection=factory.createConnection();
    			//开启连接
    			connection.start();
    			//创建 session
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//创建 Topic
    			Topic topic=session.createTopic("topic001");
    			
    			//创建消费者2- 增加 selector name='A'
    			MessageConsumer consumer2=session.createConsumer(topic, "name='A'");
    			TextMessage message2=(TextMessage)consumer2.receive();
    			System.out.println("topic获取 name='A':"+message2.getText());
    			
    		}  catch (JMSException e) { 
    			System.out.println("出现问题:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    				if(session!=null) { 
    					session.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭连接出现问题:"+e);
    			}
    		}
    		
    	}
    }

### 总结 ###

> 消息选择器可以让客户端直接消费自己想要的消息，而忽略自己不想要的消息，这样可以使用一个Queue或者Topic来完成不同场景下的工作。  
> 但是一定要注意Queue类型中，一旦一个Queue由不同的选择器条件消费者处理，其中一个的处理速率会影响另一个。  
> 如果客户端没有指定条件选择，那么对于这个消费者来说生产者端的条件设置不会有实际的作用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70]: /images/20211210/314e0039eb1c4c75a6a8c9a12c6e6c96.png