JMS两种模型的介绍和ActiveMQ的简单实例

云计算 waitig 504℃ 百度已收录 0评论

前言:

在前面的一篇文章对ActiveMQ有了一个简单了解ActiveMQ的介绍和安装,这里对JMS的两种模型进行简单介绍,并对ActiveMQ基于JMS的两种模型的实现结合代码做一个一个简单的记录,方便以后进行查看。

一、导入maven依赖

使用5.11.2版本的jar包

<!-- activemq -->
	<dependency>
		<groupId>org.apache.activemq</groupId>
		<artifactId>activemq-all</artifactId>
		<version>5.11.2</version>
	</dependency>

二、JMS的两种模型介绍

2.1P2P

涉及概念:

1.消息队列(Queue)

2.发送者(Sender)

3.接收者(Receiver)

4.每个消息都被发送到一个特定的队列,接收者从队列中获取消息。队列保留着消息,4直到他们被消费或超时。

P2P的特点

1.每个消息只有一个消费者(Consumer)(即一旦被消费,消息就不再在消息队列中)

2.发送者和接收者之间在时间上没有依赖性,也就是说当发送者发送了消息之后,不管接收者有没有正在运行,它不会影响到消息被发送到队列

3.接收者在成功接收消息之后需向队列应答成功

应用

如果你希望发送的每个消息都应该被成功处理的话,那么你需要P2P模式

2.2Pub/Sub

涉及概念:

1主题(Topic)

2发布者(Publisher)

3订阅者(Subscriber) 

4客户端将消息发送到主题。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

Pub/Sub的特点

1.每个消息可以有多个消费者

2.发布者和订阅者之间有时间上的依赖性。针对某个主题(Topic)的订阅者,它必须创建一个订阅者之后,才能消费发布者的消息,而且为了消费消息,订阅者必须保持运行的状态。

3.为了缓和这样严格的时间相关性,JMS允许订阅者创建一个可持久化的订阅。这样,即使订阅者没有被激活(运行),它也能接收到发布者的消息。

应用:

如果你希望发送的消息可以不被做任何处理、或者被一个消息者处理、或者可以被多个消费者处理的话,那么可以采用Pub/Sub模型

三、ActiveMQ对Queue和Topic的实现

3.1Queue

3.1.1Producer 生产者

@Test
	public void testQueueProducer() throws Exception {
		// 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。
		//brokerURL服务器的ip及端口号
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
		// 第二步:使用ConnectionFactory对象创建一个Connection对象。
		Connection connection = connectionFactory.createConnection();
		// 第三步:开启连接,调用Connection对象的start方法。
		connection.start();
		// 第四步:使用Connection对象创建一个Session对象。
		//第一个参数:是否开启事务。true:开启事务,第二个参数忽略。
		//第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个Queue对象。
		//参数:队列的名称。
		Queue queue = session.createQueue("test-queue");
		// 第六步:使用Session对象创建一个Producer对象。
		MessageProducer producer = session.createProducer(queue);
		// 第七步:创建一个Message对象,创建一个TextMessage对象。
		/*TextMessage message = new ActiveMQTextMessage();
		message.setText("hello activeMq,this is my first test.");*/
		TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");
		// 第八步:使用Producer对象发送消息。
		producer.send(textMessage);
		// 第九步:关闭资源。
		producer.close();
		session.close();
		connection.close();
	}

3.1.2Consumer 消费者

@Test
	public void testQueueConsumer() throws Exception {
		// 第一步:创建一个ConnectionFactory对象。
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
		// 第二步:从ConnectionFactory对象中获得一个Connection对象。
		Connection connection = connectionFactory.createConnection();
		// 第三步:开启连接。调用Connection对象的start方法。
		connection.start();
		// 第四步:使用Connection对象创建一个Session对象。
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 第五步:使用Session对象创建一个Destination对象。和发送端保持一致queue,并且队列的名称一致。
		Queue queue = session.createQueue("test-queue");
		// 第六步:使用Session对象创建一个Consumer对象。
		MessageConsumer consumer = session.createConsumer(queue);
		// 第七步:接收消息。
		consumer.setMessageListener(new MessageListener() {
			
			@Override
			public void onMessage(Message message) {
				try {
					TextMessage textMessage = (TextMessage) message;
					String text = null;
					//取消息的内容
					text = textMessage.getText();
					// 第八步:打印消息。
					System.out.println(text);
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		});
		//等待键盘输入
		System.in.read();
		// 第九步:关闭资源
		consumer.close();
		session.close();
		connection.close();

3.2Topic

3.2.1 Producer 生产者

@Test
	public void testTopicProducer() throws Exception {
		// 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。
		// brokerURL服务器的ip及端口号
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
		// 第二步:使用ConnectionFactory对象创建一个Connection对象。
		Connection connection = connectionFactory.createConnection();
		// 第三步:开启连接,调用Connection对象的start方法。
		connection.start();
		// 第四步:使用Connection对象创建一个Session对象。
		// 第一个参数:是否开启事务。true:开启事务,第二个参数忽略。
		// 第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答。
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个topic对象。
		// 参数:话题的名称。
		Topic topic = session.createTopic("test-topic");
		// 第六步:使用Session对象创建一个Producer对象。
		MessageProducer producer = session.createProducer(topic);
		// 第七步:创建一个Message对象,创建一个TextMessage对象。
		/*
		 * TextMessage message = new ActiveMQTextMessage(); message.setText(
		 * "hello activeMq,this is my first test.");
		 */
		TextMessage textMessage = session.createTextMessage("hello activeMq,this is my topic test");
		// 第八步:使用Producer对象发送消息。
		producer.send(textMessage);
		// 第九步:关闭资源。
		producer.close();
		session.close();
		connection.close();
	}

3.2.2 Consumer 消费者

@Test
	public void testTopicConsumer() throws Exception {
		// 第一步:创建一个ConnectionFactory对象。
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
		// 第二步:从ConnectionFactory对象中获得一个Connection对象。
		Connection connection = connectionFactory.createConnection();
		// 第三步:开启连接。调用Connection对象的start方法。
		connection.start();
		// 第四步:使用Connection对象创建一个Session对象。
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 第五步:使用Session对象创建一个Destination对象。和发送端保持一致topic,并且话题的名称一致。
		Topic topic = session.createTopic("test-topic");
		// 第六步:使用Session对象创建一个Consumer对象。
		MessageConsumer consumer = session.createConsumer(topic);
		// 第七步:接收消息。
		consumer.setMessageListener(new MessageListener() {

			@Override
			public void onMessage(Message message) {
				try {
					TextMessage textMessage = (TextMessage) message;
					String text = null;
					// 取消息的内容
					text = textMessage.getText();
					// 第八步:打印消息。
					System.out.println(text);
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		});
		System.out.println("topic的消费端03。。。。。");
		// 等待键盘输入
		System.in.read();
		// 第九步:关闭资源
		consumer.close();
		session.close();
		connection.close();
	}

四、测试

我们可以通过客户端对我们发送的和接受的消息进行查看


以上JMS两种模型和ActiveMQ基于JMS实现两种模型的实例,本人初学者,如果上边有什么错误的地方,请大家及时留言纠正,感激不尽,共同进步。


本文由【waitig】发表在等英博客
本文固定链接:JMS两种模型的介绍和ActiveMQ的简单实例
欢迎关注本站官方公众号,每日都有干货分享!
等英博客官方公众号
点赞 (0)分享 (0)