【Active入门-2】ActiveMQ学习-生产者与消费者

1个生产者，1个消费者，使用Queue：方式1：生产者将消息发送到Queue中，退出；

然后运行消费者：

、可以看到，可以接收到消息。方式2：先运行消费者程序：

然后运行生产者：

消费者见下图：

1个生产者，2个消费者，使用Queue先运行消费者1：


在运行消费者2：

接下来运行生产者：

下面是消费者消费情况：



：总结：1. 使用Queue时，生产者只要将Message发送到MQ服务器端，消费者就可以进行消费，而无需生产者程序一直运行；2. 消息是按照先入先出的顺序，一旦有消费者将Message消费，该Message就会从MQ服务器队列中删去；3. 有文章说，“生产者”<-->"消费者"是一对一的关系，其实并不准确，从上面可以看出，一个生产者产生的消息，可以被多个消费者进行消费，只不过多个消费者在消费消息时是竞争的关系，先得到的先消费，一旦消费完成，该消息就会出队列，就不能被其他消费者再消费了，即“一次性消费”。
特点：1.“离线一次性消费”；离线：指生产者在发送消息时，不需要消费者在线，生产者只需要将消息发送到MQ队列中，消费者可以稍后上线取消息；一次性消费：指消费者之间存在竞争关系，任何一个消费者将消息消费掉之后，其他消费者都不能再进行消费；若是只有1个生产者和1个消费者，就是我们熟悉的“点对点”通信了；
类似场景对比：类似送快递，快递员（producer）将快递（Message）放到指定地点（destination）后，就可以离开了，拿快递的人（customer）在接收到通知后，到指定地点（destination）去取快递（Message）就可以了。当然，取快递时可能要进行身份验证，这就涉及到创建连接（connection）时，需要指定用户名和密码了。还有就是，实际生活中，当快递员把快递放好之后，照理说应该通知客户去哪里取快递，而ActiveMq帮我们做好了一切，通知的工作Activemq会帮我们实现，而无需我们亲自编码通知消费者，生产者只需要将Message放到Mq中即可，通知消费者的工作，mq会帮我们处理。
消费者程序：

package com.ll.activemq;

import java.util.Collection;

import java.util.Iterator;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.MapMessage;

import javax.jms.MessageConsumer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Receiver {

public static void main(String[] args) {

// ConnectionFactory ：连接工厂，JMS 用它创建连接

ConnectionFactory connectionFactory;

// Connection ：JMS 客户端到JMS Provider 的连接

Connection connection = null;

// Session： 一个发送或接收消息的线程

Session session;

// Destination ：消息的目的地;消息发送给谁.

Destination destination;

// 消费者，消息接收者

MessageConsumer consumer;

connectionFactory = new ActiveMQConnectionFactory(

ActiveMQConnection.DEFAULT_USER,

ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");

try {

// 构造从工厂得到连接对象

connection = connectionFactory.createConnection();

// 启动

connection.start();

// 获取操作连接

session = connection.createSession(Boolean.FALSE,

Session.AUTO_ACKNOWLEDGE);

// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置

destination = session.createQueue("FirstQueue");

consumer = session.createConsumer(destination);

// consumer2 = session.createConsumer(destination);

System.out.println("消费者1：消费者程序开始运行...");

while (true) {

// 设置接收者接收消息的时间，为了便于测试，这里谁定为100s

TextMessage message = (TextMessage) consumer.receive(18000);

// Object message = (Object) consumer.receive(28000);

if (null != message) {

System.out.println("消费者1：收到消息-->" + message.getText());

} else {

System.out.println("消费者1：运行结束...\n");

break;

}

}

} catch (Exception e) {

e.printStackTrace();

} finally {

	try {

if (null != connection)

connection.close();

} catch (Throwable ignore) {

}

}

}

}

[/code]生产者程序：

package com.ll.activemq;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Sender {

private static final int SEND_NUMBER = 5;

public static void main(String[] args) {

// ConnectionFactory ：连接工厂，JMS 用它创建连接

ConnectionFactory connectionFactory;

// Connection ：JMS 客户端到JMS Provider 的连接

Connection connection = null;

// Session： 一个发送或接收消息的线程

Session session;

// Destination ：消息的目的地;消息发送给谁.

Destination destination;

// MessageProducer：消息发送者

MessageProducer producer;

// TextMessage message;

// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar

connectionFactory = new ActiveMQConnectionFactory(

ActiveMQConnection.DEFAULT_USER,

ActiveMQConnection.DEFAULT_PASSWORD,

"tcp://localhost:61616");

try {

// 构造从工厂得到连接对象

connection = connectionFactory.createConnection();

// 启动

connection.start();

// 获取操作连接

session = connection.createSession(Boolean.TRUE,

Session.AUTO_ACKNOWLEDGE);

// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置

destination = session.createQueue("FirstQueue");

// 得到消息生成者【发送者】

producer = session.createProducer(destination);

// 设置不持久化，此处学习，实际根据项目决定

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

// 构造消息，此处写死，项目就是参数，或者方法获取

sendMessage(session, producer);

session.commit();

} catch (Exception e) {

e.printStackTrace();

} finally {

    try {

if (null != connection)

connection.close();

} catch (Throwable ignore) {

}

}

}

public static void sendMessage(Session session, MessageProducer producer)

throws Exception {

for (int i = 1; i <= SEND_NUMBER; i++) {

TextMessage message = session

.createTextMessage("ActiveMq 发送的消息" + i);

// 发送消息到目的地方

System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);

producer.send(message);

}

System.out.println("生产者程序退出...");

}

}

[/code]

来自为知笔记(Wiz)

附件列表