ActiveMQ的简单Topic实现案例

红太狼 2022-09-24 13:24 167阅读 0赞

首先我们在这里要说一下消息中间件中有两个角色，生产者Producer与消费者Consumer，简单理解即使发发送消息者与接收消息者，在编写代码之前我们需要将下载到的ActiveMQ压缩包中的activemq-all-5.13.4.jar包导入项目，接下来我们用代码展示一下ActiveMQ的独特魅力。

Producer类:

package com.yc.mytest;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class MyProducer {
    	private static final String URL=ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址
    	private static String TOPIC_NAME="MyTopicTest"; //topic名字
    	
    	public static void main(String[] args) {
    		ConnectionFactory factory; //连接工厂
    		Connection conn = null; //连接
    		Session session; //会话
    		Destination dest; //消息的目的地
    		MessageProducer producer; //消息生产者
    
    		try {
    			//初始化连接工厂
    			factory=new ActiveMQConnectionFactory(URL); 
    
    			//获取连接
    			conn=factory.createConnection(); 
    
    			//开始连接
    			conn.start();
    
    			//创建Session，此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
    			session=conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			
    			//创建topic,producer和consumer将根据TOPIC_NAME来发送/接收对应的消息 
    			dest=session.createTopic(TOPIC_NAME); 
    			
    			//创建生产者
    			producer=session.createProducer(dest);
    			
    			//发送消息
    			sendMessage(session, producer);
    		} catch (Exception e) {
    			System.out.println("出错啦!!!");
    			e.printStackTrace();
    		} finally{
    			if(conn!=null){
    				try {
    					//关闭连接
    					conn.close();
    				} catch (JMSException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	private static void sendMessage(Session session,MessageProducer producer) throws JMSException{
            //初始化一个mq消息
            TextMessage message = session.createTextMessage() ;//= session.createTextMessage("你好 ActiveMq " + i);
           
            //填写消息内容
            message.setText("你好 ActiveMq");
            
            //发送消息
            producer.send(message);
            
            System.out.println("消息发送成功!!!");
    	}
    }

Consumer类:

package com.yc.mytest;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class MyConsumer implements MessageListener {
    	private static final String URL=ActiveMQConnection.DEFAULT_BROKER_URL; //默认的连接地址
    	private static String TOPIC_NAME="MyTopicTest"; //topic名字
    
    	public static void main(String[] args) {
    		ConnectionFactory factory; //连接工厂
    		Connection conn = null; //连接
    		Session session; //会话
    		Destination dest; //消息的目的地
    		MessageConsumer consumer; //消费者
    		
    		try {
    			//初始化连接工厂
    			factory=new ActiveMQConnectionFactory(URL); 
    
    			//获取连接
    			conn=factory.createConnection(); 
    			
    			//开始连接
    			conn.start(); 
    			
    			//创建Session，此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
    			session=conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			
    			//创建topic,producer和consumer将根据TOPIC_NAME来发送/接收对应的消息 
    			dest=session.createTopic(TOPIC_NAME); 
    			
    			//创建消费者
    			consumer=session.createConsumer(dest);
    			
    			//初始化MessageListener
    			MyConsumer me=new MyConsumer();
    			
    			//给消费者设定监听对象
    			consumer.setMessageListener(me);
    		} catch (Exception e) {
    			System.out.println("出错啦!!!");
    			e.printStackTrace();
    		}
    	}
    	
    	public void onMessage(Message message) {
    		TextMessage txtMessage = (TextMessage)message;
    		String text;
    		try {
    			text=txtMessage.getText();
    			//txtMessage.acknowledge(); //通知服务器
    			System.out.println("收到消息:"+text);
    		} catch (JMSException e) {
    			System.out.println("出错啦!!!");
    			e.printStackTrace();
    		}
    	}
    }

好，到现在，我们的代码就实现了。0.0简单吧，我们来运行一下，在topic模式下需要先启动消费者，在这里我们启动两个消费者来模拟群发

![Center][]

启动生产者发送信息

![Center 1][]

在去看两个消费者，会收到同样的信息

![Center 2][]![Center 3][]

我们去监控中看下信息

![Center 4][]

好，到此，我们最简单的Topic模式就已经实现了![大笑][laugh.gif]

[Center]: /images/20220719/d0bc38ada61940838660d8749372bed6.png
[Center 1]: /images/20220719/ef06aec3ac40405082a1efba5040feae.png
[Center 2]: https://img-blog.csdn.net/20160801150016351?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 3]: /images/20220719/7773e2fed3a347a2901c5936fd280fbe.png
[Center 4]: /images/20220719/e85e7b49d05741f8b4bb9e3c429b28e3.png
[laugh.gif]: /images/20220719/0dd7e946a3bc4f848f8b3980401418c5.png