Java实现

添加相应的jar包

<dependency>

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-all</artifactId>

<version>x.xx.x</version>

</dependency>

创建生产者类（点对点）

public class ProducerTest {

// 异步发送asyn

// 死信队列 DLQ

// 文件上传

// header properties使用

// jdbc存储

// byteMsg objMsg inputMsg

// mq ptp 的使用场景

// mq中所有的队列名以及每个队列中未被消费的消息数量

// mq sub/pub 的使用场景

@Test

public void testQueueProducer() throws JMSException{

// 第一步：创建ConnectionFactory对象，需要指定服务端ip及端口号。

//brokerURL服务器的ip及端口号

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.114.129: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);

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

// 第七步：创建一个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.send(textMessage);

producer.close();

session.close();

connection.close();

}

}

创建消费者类（点对点）

public class ConsumerTest {

@Test

public void testQueueConsumer() throws JMSException, IOException{

// 第一步：创建一个ConnectionFactory对象。

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.114.129: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("queue.stu");

// 第六步：使用Session对象创建一个Consumer对象。

MessageConsumer consumer = session.createConsumer(queue);

// 第七步：接收消息。

consumer.setMessageListener(new MessageListener()

@Override

public void onMessage(Message message) {

try {

//if (TextMessage.class.isAssignableFrom(message.getClass()))

if (message instanceof TextMessage) {

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();

}

}

运行active服务器验证实现情况

发布（对点对点代码修改 创建Topic Destination）

// 第五步：使用Session对象创建一个Destination对象（topic、queue），此处创建一个topic对象。

// 参数：话题的名称。

Topic topic = session.createTopic("test-topic");

// 第六步：使用Session对象创建一个Producer对象。

MessageProducer producer = session.createProducer(topic);

订阅（对点对点代码修改 创建Topic Destination）

// 第五步：使用Session对象创建一个Destination对象。和发送端保持一致topic，并且话题的名称一致。

Topic topic = session.createTopic("test-topic");

// 第六步：使用Session对象创建一个Consumer对象。

MessageConsumer consumer = session.createConsumer(topic);