标签：pid cte tcp connect enc ups apach package throw
实现点对点通讯模式
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
package com.wn.ddd;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Producter {
public static void main(String[] args) throws JMSException {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
// JMS 客户端到JMS Provider 的连接
Connection connection = connectionFactory.createConnection();
//启动连接
connection.start();
// Session： 一个发送或接收消息的线程 false:代表不带事务的session AUTO_ACKNOWLEDGE：代表自动签收
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// Destination ：消息的目的地;消息发送给谁.
// 获取session注意参数值my-queue是Query的名字
Queue queue = session.createQueue("my-queue");
// MessageProducer：创建消息生产者
MessageProducer producer = session.createProducer(queue);
// 设置不持久化 PERSISTENT:代表持久化 NON_PERSISTENT：代表不持久化
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 发送消息
for (int i = 1; i <= 5; i++) {
sendMsg(session, producer, i);
}
System.out.println("发送成功！");
session.close();
connection.close();
}
/**
* 在指定的会话上，通过指定的消息生产者发出一条消息
*
* @param session
* 消息会话
* @param producer
* 消息生产者
*/
public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
// 创建一条文本消息
TextMessage message = session.createTextMessage("Hello ActiveMQ！" + i);
// 通过消息生产者发出消息
producer.send(message);
}
}
package com.wn.ddd;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class JmsReceiver {
public static void main(String[] args) throws JMSException {
// ConnectionFactory ：连接工厂，JMS 用它创建连接
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
// JMS 客户端到JMS Provider 的连接
Connection connection = connectionFactory.createConnection();
connection.start();
// Session： 一个发送或接收消息的线程 true：表单开启事务 AUTO_ACKNOWLEDGE：代表自动签收
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// Destination ：消息的目的地;消息发送给谁.
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
Queue queue = session.createQueue("my-queue");
// 消费者，消息接收者
MessageConsumer consumer = session.createConsumer(queue);
while (true) {
//receive():获取消息
TextMessage message = (TextMessage) consumer.receive();
if (null != message) {
System.out.println("收到消息：" + message.getText());
} else {
break;
}
}
//回收资源
session.close();
connection.close();
}
}
控制台执行完成后，去查看一下队列的情况
可以再次查看队列的情况
ActiveMQ消息签收机制：
客户端成功接收一条消息的标志是一条消息被签收，成功应答。
消息的签收情况分为两种：
1.带事务的session
如果session带有事务，并且事务成功提交，则消息被自动签收。如果事务回滚，则消息会被再次传递。
2.不带事务的session
不带事务的session签收方式，取决于session的配置。
ActiveMQ支持以下三种模式：
1.Session.AUTO_ACKNOWLEDGE 消息自动签收
2.Session.CLIENT_ACKNOWLEDGE 客户端调用acknowledge方法手动签收
textMessage.acknowledge() 手动签收
3.Session.DUPS_OK_ACKNOWLEDGE 不是必须签收，消息可能会重复发送。在第二次重新传送消息的时候，消息只有在被确认之后，才认为已经被成功地消费了。
package com.wn.fndy;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
//消费者
public class TopReceiver {
private static String BROKERURL = "tcp://127.0.0.1:61616";
private static String TOPIC = "my-topic";
public static void main(String[] args) throws JMSException {
start();
}
private static void start() throws JMSException {
System.out.println("消费者启动......");
// 创建ActiveMQConnectionFactory 会话工厂
ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);
Connection connection = activeMQConnectionFactory.createConnection();
// 启动JMS 连接
connection.start();
// 不开消息启事物，消息主要发送消费者,则表示消息已经签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建一个队列
Topic topic = session.createTopic(TOPIC);
MessageConsumer consumer = session.createConsumer(topic);
// consumer.setMessageListener(new MsgListener());
while (true) {
TextMessage textMessage = (TextMessage) consumer.receive();
if (textMessage != null) {
System.out.println("接受到消息:" + textMessage.getText());
// textMessage.acknowledge();// 手动签收
// session.commit();
} else {
break;
}
}
connection.close();
}
}
package com.wn.fndy;
import com.sun.scenario.effect.impl.sw.sse.SSEBlend_SRC_OUTPeer;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
//生产方
public class TOPSend {
// tcp 地址
public static final String BROKERURL = "tcp://localhost:61616";
// 目标，在ActiveMQ管理员控制台创建 http://localhost:8161/admin/queues.jsp
public static final String TOPIC = "my-topic";
public static void main(String[] args) throws JMSException {
start();
}
private static void start() throws JMSException {
System.out.println("生产者已经启动......");
//创建ActiveMQConnectionFactory会话工厂
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);
Connection connection = activeMQConnectionFactory.createConnection();
//启动JMS连接
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(null);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
send(producer,session);
System.out.println("发送成功！");
connection.close();
}
private static void send(MessageProducer producer, Session session) throws JMSException {
for (int i=1;i<=5;i++){
System.out.println("我是消息"+i);
TextMessage textMessage = session.createTextMessage("我是消息" + i);
Destination destination = session.createTopic(TOPIC);
producer.send(destination,textMessage);
}
}
}
3.1 启动两个消费者，因为只有消费者先订阅，生产者才会发送消息，消费者才能接收消息；（发布/订阅是一个发送者，对应多个消费者）
3.2 启动生产者
查看一个控制台中两个消费者的情况
标签：pid cte tcp connect enc ups apach package throw
原文地址：https://www.cnblogs.com/mayuan01/p/12391357.html
