【java-activemq】 activemq发布订阅模式
博客专区 > _______- 的博客 > 博客详情
【java-activemq】 activemq发布订阅模式
_______- 发表于5个月前
【java-activemq】 activemq发布订阅模式
  • 发表于 5个月前
  • 阅读 8
  • 收藏 0
  • 点赞 0
  • 评论 0

新睿云服务器60天免费使用,快来体验!>>>   

Destination destination = session.createTopic("xxxx");

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicPublisher {
	private static String user = ActiveMQConnection.DEFAULT_USER;  
	private static String password =ActiveMQConnection.DEFAULT_PASSWORD;  
	private static String url =  "tcp://localhost:61616";  

	public static void main(String[] args)throws Exception {  
		// ConnectionFactory :连接工厂,JMS 用它创建连接  
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);  
		// Connection :JMS 客户端到JMS Provider 的连接  
		Connection connection = connectionFactory.createConnection();  
		// Connection 启动  
		connection.start();  
		System.out.println("Connection is start...");  
		// Session: 一个发送或接收消息的线程  
		Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);  
		// Topicr :消息的目的地;消息发送给谁.  
		Topic  topic = session.createTopic("example.A");  
		// MessageProducer:消息发送者  
		MessageProducer producer = session.createProducer(topic);  
		// 设置不持久化,此处学习,实际根据项目决定  
		producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
		// 构造消息,此处写死,项目就是参数,或者方法获取  
		sendMessage(session, producer);  
		session.commit();  

		connection.close();  
		System.out.println("send text ok.");  
	}  

	public static void sendMessage(Session session, MessageProducer producer)  throws Exception {  
		for (int i = 1; i <= 100; i++) {//有限制,达到1000就不行  
			TextMessage message = session.createTextMessage("ActiveMq 发送的消息" + i);  
			// 发送消息到目的地方  
			System.out.println("发送消息:" + "ActiveMq 发送的消息" + i);  
			producer.send(message);  
		}  
	}  
}


import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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 javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicSubscriber{
	private static String user = ActiveMQConnection.DEFAULT_USER;  
	private static String password =ActiveMQConnection.DEFAULT_PASSWORD;  
	private static String url = "tcp://localhost:61616";  
	private static boolean flag=true;

	public static void main(String[] args) throws Exception{  
		// ConnectionFactory :连接工厂,JMS 用它创建连接  
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);  
		// Connection :JMS 客户端到JMS Provider 的连接  
		Connection connection = connectionFactory.createConnection();  
		connection.start();  
		// Session: 一个发送或接收消息的线程  
		final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
		// Destination :消息的目的地;消息发送给谁.  
		Topic topic=session.createTopic("example.B");  
		// 消费者,消息接收者  
		MessageConsumer consumer = session.createConsumer(topic);  
		consumer.setMessageListener(new MessageListener(){//有事务限制  
			@Override  
			public void onMessage(Message message) {  
				try {  
					TextMessage textMessage=(TextMessage)message;  
					System.out.println(textMessage.getText());  
				} catch (JMSException e1) {  
					e1.printStackTrace();  
				}  
				try {  
					session.commit();  
				} catch (JMSException e) {  
					e.printStackTrace();  
				}  
			}  
		});  

		//		  另外一种接受方式 
		//		while(flag){  
		//			TextMessage message = (TextMessage)consumer.receive(1);  
		//			if(message != null){  
		//				if("stop".equals( message.getText())){  
		//					flag = false;  
		//				}  
		//			}  
		//		}  
	}
}


import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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 javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class TopicSubscriber2{
	private static String user = ActiveMQConnection.DEFAULT_USER;  
	private static String password =ActiveMQConnection.DEFAULT_PASSWORD;  
	private static String url = "tcp://localhost:61616";  
	private static boolean flag=true;

	public static void main(String[] args) throws Exception{  
		// ConnectionFactory :连接工厂,JMS 用它创建连接  
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);  
		// Connection :JMS 客户端到JMS Provider 的连接  
		Connection connection = connectionFactory.createConnection();  
		connection.start();  
		// Session: 一个发送或接收消息的线程  
		final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);  
		// Destination :消息的目的地;消息发送给谁.  
		Topic topic=session.createTopic("example.A");  
		// 消费者,消息接收者  
		MessageConsumer consumer = session.createConsumer(topic);  
		consumer.setMessageListener(new MessageListener(){//有事务限制  
			@Override  
			public void onMessage(Message message) {  
				try {  
					TextMessage textMessage=(TextMessage)message;  
					System.out.println(textMessage.getText());  
				} catch (JMSException e1) {  
					e1.printStackTrace();  
				}  
				try {  
					session.commit();  
				} catch (JMSException e) {  
					e.printStackTrace();  
				}  
			}  
		});  

		//		  另外一种接受方式 
		//		while(flag){  
		//			TextMessage message = (TextMessage)consumer.receive(1);  
		//			if(message != null){  
		//				if("stop".equals( message.getText())){  
		//					flag = false;  
		//				}  
		//			}  
		//		}  
	}
}

  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
粉丝 3
博文 73
码字总数 34982
×
_______-
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: