ActiveMQ点对点消息传递

快来打我* 2022-02-02 04:41 242阅读 0赞

## ActiveMQ ##

## 处理对象消息 ##

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

package com.sxt.bean;
    
    import java.io.Serializable;
    import java.util.Date;
    /**
     * 自定义对象
     * @author xn
     *
     */
    
    
    public class User implements Serializable{
    
    	private static final long serialVersionUID = 1L;
    	private String id;
    	
    	private String name;
    	
    	private Date datatime;
    
    	public Date getDatatime() {
    		return datatime;
    	}
    
    	public void setDatatime(Date datatime) {
    		this.datatime = datatime;
    	}
    
    	@Override
    	public String toString() {
    		return "User [id=" + id + ", name=" + name + ", datatime=" + datatime + "]";
    	}
    
    	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 User(String id, String name, Date datatime) {
    		super();
    		this.id = id;
    		this.name = name;
    		this.datatime = datatime;
    	}
    }

## 定义消息生产者 ##

package com.sxt.bean;
    
    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.Message;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQMessage;
    import org.apache.activemq.command.ActiveMQObjectMessage;
    import org.apache.activemq.command.ActiveMQTextMessage;
    
    /**
     * 定义生产者
     * @author Administrator
     *
     */
    public class ActivemqObjectProducer {
    	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 {
    		// 连接工厂，用于创建Connection对象
    		ConnectionFactory factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
    		//创建Connection对象
    		Connection connection = factory.createConnection();
    		//启动连接对象
    		connection.start();
    		
    		/*
    		 * 创建会话对象 
    		 * 方法 - connection.createSession(boolean transacted, int acknowledgeMode); 
    		 * transacted - 是否使用事务, 
    		 * 可选值为true|false 
    		 * true - 使用事务, 当设置此变量值, 则acknowledgeMode参数无效, 
    		 * 建议传递的acknowledgeMode参数值为 Session.SESSION_TRANSACTED 
    		 * false - 不使用事务, 设置此变量值,则acknowledgeMode参数必须设置. 
    		 * acknowledgeMode - 消息确认机制, 可选值为:
    		 * Session.AUTO_ACKNOWLEDGE - 自动确认消息机制 
    		 * Session.CLIENT_ACKNOWLEDGE -客户端确认消息机制 
    		 * Session.DUPS_OK_ACKNOWLEDGE - 有副本的客户端确认消息机制
    		 */
    				Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    				// 创建目的地，目的地的命名既是队列的命令
    				Destination destination = session.createQueue("Object-MQ");
    				// 创建消息生成者, 创建的消息生成者与某目的地对应, 即方法参数目的地
    				MessageProducer producer = session.createProducer(destination);
    				// 创建消息对象，创建一个文本消息对象。此消息对象中保存要传递的文本数据.
    				ActiveMQObjectMessage message = new ActiveMQObjectMessage();
    				message.setObject(new User("2", "lisi", new Date()));
    				//发送消息
    				producer.send(message);
    				//关闭连接
    				producer.close();
    				session.close();
    				connection.close();
    	}
    }

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

## 定义消息消费者 ##

package com.sxt.bean;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.ObjectMessage;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 定义消费者
     * @author xn
     *
     */
    public class ActivemqObjectConsumer {
    	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 {
    		// 连接工厂，用于创建Connection对象
    		ConnectionFactory factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
    		//创建Connection对象
    		Connection connection = factory.createConnection();
    		//启动连接对象
    		connection.start();
    		
    		/*
    		 * 创建会话对象 
    		 * 方法 - connection.createSession(boolean transacted, int acknowledgeMode); 
    		 * transacted - 是否使用事务, 
    		 * 可选值为true|false 
    		 * true - 使用事务, 当设置此变量值, 则acknowledgeMode参数无效, 
    		 * 建议传递的acknowledgeMode参数值为 Session.SESSION_TRANSACTED 
    		 * false - 不使用事务, 设置此变量值,则acknowledgeMode参数必须设置. 
    		 * acknowledgeMode - 消息确认机制, 可选值为:
    		 * Session.AUTO_ACKNOWLEDGE - 自动确认消息机制 
    		 * Session.CLIENT_ACKNOWLEDGE -客户端确认消息机制 
    		 * Session.DUPS_OK_ACKNOWLEDGE - 有副本的客户端确认消息机制
    		 */
    				Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    				// 创建目的地，目的地的命名既是队列的命令
    				Destination destination = session.createQueue("Object-MQ");
    				// 创建消息消费者, 创建的消息消费者与某目的地对应, 即方法参数目的地.
    				MessageConsumer consumer = session.createConsumer(destination);
    				// 从ActiveMQ中获取消息
    				Message message = consumer.receive();
    				//打印消息
    				System.out.println("获取的消息是："+((ObjectMessage)message).getObject());
    				//关闭连接
    				consumer.close();
    				session.close();
    				connection.close();
    }
    }

![在这里插入图片描述][2019051016554983.png]  
![在这里插入图片描述][20190510165559394.png]

## 观察者模式 ##

## 事件源 ##

事件发生的源头。监听器监听的具体位置

## 事件 ##

具体触发的事件。如：单击事件等  
其中必然包含事件源信息

## 监听器 ##

处理事件的代码逻辑

## 定义生产者代码 ##

package com.dpb.observe;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * ActiveMQ中的生产者(Producer)
     * @author dengp
     *
     */
    public class MyProducer {
    
    	public void sendhello2ActiveMq(String messageText) {
    		// 连接工厂，用于创建Connection对象
    		ConnectionFactory factory = null;
    		// activeMQ 连接对象
    		Connection conn = null;
    		// 一次和ActiveMQ的持久会话对象
    		Session session = null;
    		// 目的地
    		Destination destination = null;
    		// 消息发送者
    		MessageProducer producer = null;
    		// 封装消息的对象
    		Message message = null;
    		try {
    			/*
    			 * 创建链接工厂 ActiveMQConnectionFactory -由ActiveMQ实现的ConnectionFactory接口实现类. 
    			 * 构造方法: public ActiveMQConnectionFactory(String userName, String password,
    			 * String brokerURL) 
    			 * userName - 访问ActiveMQ服务的用户名,用户名可以通过jetty-realm.properties配置文件配置. 
    			 * password - 访问ActiveMQ服务的密码,密码可以通过jetty-realm.properties配置文件配置. 
    			 * brokerURL -访问ActiveMQ服务的路径地址. 路径结构为 - 协议名://主机地址:端口号 此链接基于TCP/IP协议.
    			 */
    			factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.88.121:61616");
    			// 创建链接对象
    			conn = factory.createConnection();
    			// 启动连接对象
    			conn.start();
    			/*
    			 * 创建会话对象 
    			 * 方法 - connection.createSession(boolean transacted, int acknowledgeMode); 
    			 * transacted - 是否使用事务, 
    			 * 可选值为true|false 
    			 * true - 使用事务, 当设置此变量值, 则acknowledgeMode参数无效, 
    			 * 建议传递的acknowledgeMode参数值为 Session.SESSION_TRANSACTED 
    			 * false - 不使用事务, 设置此变量值,则acknowledgeMode参数必须设置. 
    			 * acknowledgeMode - 消息确认机制, 可选值为:
    			 * Session.AUTO_ACKNOWLEDGE - 自动确认消息机制 
    			 * Session.CLIENT_ACKNOWLEDGE -客户端确认消息机制 
    			 * Session.DUPS_OK_ACKNOWLEDGE - 有副本的客户端确认消息机制
    			 */
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			// 创建目的地，目的地的命名既是队列的命令
    			destination = session.createQueue("MQ-Hello-observe");
    			// 创建消息生成者, 创建的消息生成者与某目的地对应, 即方法参数目的地.
    			producer = session.createProducer(destination);
    			// 创建消息对象，创建一个文本消息对象。此消息对象中保存要传递的文本数据.
    			message = session.createTextMessage(messageText);
    
    			// 发送消息
    			producer.send(message);
    		} catch (Exception e) {
    			e.printStackTrace();
    			System.out.println("访问ActiveMQ服务发生错误!!");
    		} finally {
    			try {
    				// 回收消息发送者资源
    				if (null != producer)
    					producer.close();
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    			try {
    				// 回收会话资源
    				if (null != session)
    					session.close();
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    			try {
    				// 回收链接资源
    				if (null != conn)
    					conn.close();
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

## 定义消费者模式 ##

package com.dpb.observe;
    
    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.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * ActiveMQ中的消费者(Consumer)
     * @author dengp
     *
     */
    public class MyConsumer {
    
    	public void reciveHelloFormActiveMq() {
    		// 连接工厂，用于创建Connection对象
    		ConnectionFactory factory = null;
    		// activeMQ 连接对象
    		Connection conn = null;
    		// 一次和ActiveMQ的持久会话对象
    		Session session = null;
    		// 目的地
    		Destination destination = null;
    		// 消息消费者
    		MessageConsumer consumer = null;
    		// 封装消息的对象
    		Message message = null;
    		try {
    			/*
    			 * 创建链接工厂 ActiveMQConnectionFactory -由ActiveMQ实现的ConnectionFactory接口实现类. 
    			 * 构造方法: public ActiveMQConnectionFactory(String userName, String password,
    			 * String brokerURL) 
    			 * userName - 访问ActiveMQ服务的用户名,用户名可以通过jetty-realm.properties配置文件配置. 
    			 * password - 访问ActiveMQ服务的密码,密码可以通过jetty-realm.properties配置文件配置. 
    			 * brokerURL -访问ActiveMQ服务的路径地址. 路径结构为 - 协议名://主机地址:端口号 此链接基于TCP/IP协议.
    			 */
    			factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.88.121:61616");
    			// 创建链接对象
    			conn = factory.createConnection();
    			// 启动连接对象
    			conn.start();
    			/*
    			 * 创建会话对象 
    			 * 方法 - connection.createSession(boolean transacted, int acknowledgeMode); 
    			 * transacted - 是否使用事务, 
    			 * 可选值为true|false 
    			 * true - 使用事务, 当设置此变量值, 则acknowledgeMode参数无效, 
    			 * 建议传递的acknowledgeMode参数值为 Session.SESSION_TRANSACTED 
    			 * false - 不使用事务, 设置此变量值,则acknowledgeMode参数必须设置. 
    			 * acknowledgeMode - 消息确认机制, 可选值为:
    			 * Session.AUTO_ACKNOWLEDGE - 自动确认消息机制 
    			 * Session.CLIENT_ACKNOWLEDGE -客户端确认消息机制 
    			 * Session.DUPS_OK_ACKNOWLEDGE - 有副本的客户端确认消息机制
    			 */
    			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			// 创建目的地，目的地的命名既是队列的命令
    			destination = session.createQueue("MQ-Hello-observe");
    			// 创建消息消费者, 创建的消息消费者与某目的地对应, 即方法参数目的地.
    			consumer = session.createConsumer(destination);
    			// 监听ActiveMQ服务中的消息,当发现消息的时候,自动处理
    			consumer.setMessageListener(new MessageListener() {
    				/**
    				 * 当用消息到来的时候触发该方法，在该方法中处理消息
    				 */
    				@Override
    				public void onMessage(Message message) {
    					TextMessage textMessage = (TextMessage) message;
    					String messageString = null;
    					try {
    						messageString = textMessage.getText();
    					} catch (JMSException e) {
    						e.printStackTrace();
    						messageString = "处理消息失败!!";
    					}
    					System.out.println("处理的消息内容是 : " + messageString);
    				}
    			});
    			// 阻塞进程
    			System.in.read();
    		} catch (Exception e) {
    			e.printStackTrace();
    			System.out.println("访问ActiveMQ服务发生错误!!");
    		} finally {
    			try {
    				// 回收消息发送者资源
    				if (null != consumer)
    					consumer.close();
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    			try {
    				// 回收会话资源
    				if (null != session)
    					session.close();
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    			try {
    				// 回收链接资源
    				if (null != conn)
    					conn.close();
    			} catch (JMSException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

## 测试 ##

生产者

public static void main(String[] args) {
    	MyProducer pro = new MyProducer();
    	pro.sendhello2ActiveMq("你好啊...listener");
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 1]  
消费者

public static void main(String[] args) {
    	MyConsumer con = new MyConsumer();
    	con.reciveHelloFormActiveMq();
    }

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70]: /images/20220202/ed2950f415c749cb8d8bd50e2729e802.png
[2019051016554983.png]: /images/20220202/d408682617e04c4696e52c516b2987cd.png
[20190510165559394.png]: /images/20220202/6ba57aa4e92d43909f5eb7aab1449397.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 1]: /images/20220202/7cd967a7141f4653a3bdae37fcb8964a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwOTg0NTYx_size_16_color_FFFFFF_t_70 2]: /images/20220202/3bca8f3bb8674b99a937a20083c90e67.png