MQ初认识
程序员文章站
2022-03-05 11:37:35
...
1.两大目的地
a.在点对点的消息传递域中，目的地被称为队列（queue）。
//生产者
public class JmsProduce {
public static final String URL = "tcp://10.112.70.211:61616";
public static final String queue_name = "queue01";
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue(queue_name);
MessageProducer producer = session.createProducer(queue);
for (int i = 1; i < 3; i++) {
TextMessage textMessage = session.createTextMessage("mag..." + i);
producer.send(textMessage);
}
producer.close();
session.close();
connection.close();
System.out.println("....");
}
}
生产者
//方式一：同步阻塞方式receive()
//订阅者调用MessageConsumer的receive()来接受消息，receive()能在接受消息之前一直被阻塞。
public class JmsConsumer {
public static final String URL = "tcp://10.112.70.211:61616";
public static final String queue_name = "queue";
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue(queue_name);
MessageConsumer consumer = session.createConsumer(queue);
while (true){
TextMessage receive = (TextMessage) consumer.receive();
if (null != receive){
System.out.println("消息:"+receive.getText());
}
else {
System.out.println("结束");
break;
}
}
consumer.close();
session.close();
connection.close();
System.out.println("完成....");
}
}
//方式二：设置监听方式(异步非阻塞方式onMessage())
//订阅者或接受者通过MessageConsumer的setMessageListener(MessageListener Listener) 注册一个消息监听器
//当消息到达之后，系统会自动调用监听器MessageListener的onMessage(message)方法
public class JmsConsumer {
public static final String URL = "tcp://10.112.70.211:61616";
public static final String queue_name = "queue01";
public static void main(String[] args) throws JMSException, IOException, InterruptedException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue(queue_name);
MessageConsumer consumer = session.createConsumer(queue);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage) message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
// System.in.read();
// Thread.sleep(1000);
consumer.close();
session.close();
connection.close();
System.out.println("完成....");
}
}
b.在发布订阅消息传递域中，目的地被称为主题（topic）。
//1.发布主题生产者
public class JmsProduce_Topic {
public static final String URL = "tcp://10.112.70.211:61616";
public static final String topic_name = "topic01";
public static void main(String[] args) throws JMSException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic(topic_name);
MessageProducer producer = session.createProducer(topic);
for (int i = 1; i < 3; i++) {
TextMessage textMessage = session.createTextMessage("Topic mag..." + i);
producer.send(textMessage);
}
producer.close();
session.close();
connection.close();
System.out.println("完成....");
}
}
//2.订阅主题消费者
public class JmsConsumer_Topic {
public static final String URL = "tcp://10.112.70.211:61616";
public static final String topic_name = "topic01";
public static void main(String[] args) throws JMSException, IOException, InterruptedException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(URL);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic(topic_name);
MessageConsumer consumer = session.createConsumer(topic);
// while (true){
// TextMessage receive = (TextMessage) consumer.receive();
// if (null != receive){
// System.out.println("消息:"+receive.getText());
// }
// else {
// System.out.println("结束");
// break;
// }
// }
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("消费者接收到消息："+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
// System.in.read();
// Thread.sleep(1000);
consumer.close();
session.close();
connection.close();
System.out.println("完成....");
}
}
2.JMS的组成和特点
JMS是消息通信规范，MQ是基于这种规范而实现的。
2.1JMS privider：
实现JMS接口与规范的消息中间件，也就是我们的MQ服务器
2.2JMS producer：
消息生产者，创建与发送JMS消息的客户端应用
2.3JMS consumer：
消息的消费者，接受与处理JMS消息的客户端应用
2.4JMS message：
消息头:
JMSdesination:消息发送的目的地，主要是指Queue与Topic
JMSDeliveryMode：持久性（服务器故障消息不丢失）和非持久性模式
JMSExpiration:设置消息过期时间，默认不过期
JMSPriority:
JMSMessageID:唯一识别每个消息的表示，由MQ产生
消息体：
TextMessage:普通的字符串消息，包含一个string
MapMessage:一个map类型的消息，key为stirng类型，而value 为java的基本类型
BytesMessage:二进制数组消息，包含一个byte[]
StreamMessage:java数据流消息，用标准流操作来顺序的填充或读取
ObjectMessage:对象消息，包含一个可序列化的java对象
3.JMS可靠性
3.1 persistent：持久性
非持久的topic 没有意义，因为发布订阅模式是先启动订阅在启动生成，消息已经被消费了。如果先启动生产者后启动订阅者，消息会被当作废消息
3.2 transaction:事务
producer提交时的事务
false:
只要执行send就进入到队列中session.createSession(false,);自动提交到mq
true:
先执行send再执行commit，消息才被真正的提交到队列中;
------------------------
关闭之前需要commit,这是在生产者的事务，负责消息的提交不出错，如果出错则回滚；
事务偏生产者/签收偏消费者
3.3 Acknowledge:签收
3.3.1非事务
自动签收：Session.AUTO_ACKNOWLEDGE
手动签收：Session.CLIENT_ACKNOWLEDGE
3.3.2 事务
消费者事务开启，只有commit后才能将全部消息变为已消费，写ack没有作用，即使写了签收也不会签收出队。
--------------签收和事务的关系-----------
a.在事务性会话中，当一个事务被成功提交则消息被自动签收。如果事务回滚，则消息会被再次传送。
b.非事务会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement）
4.Spring整合mq
4.1 maven修改，需要添加spring支持的jms的包
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.rocket</groupId>
<artifactId>test_activemq</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.targer>1.8</maven.compiler.targer>
</properties>
<dependencies>
<!--activemq需要的jar包-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.9.5</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.3.23.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.15.9</version>
</dependency>
<!--spring需要的jar-->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>4.3.23.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>4.3.23.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId>
<version>4.3.23.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-orm</artifactId>
<version>4.3.23.RELEASE</version>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjrt</artifactId>
<version>1.6.1</version>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<version>1.6.8</version>
</dependency>
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib</artifactId>
<version>2.1_2</version>
</dependency>
<!--下面是junit/log4等通用配置-->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.3</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.18</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
</dependencies>
</project>
4.2.spring配置文件
applicationContext.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="com.atguigu.activemq"></context:component-scan>
<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://10.112.70.211:61616"/>
</bean>
</property>
<property name="maxConnections" value="100"></property>
</bean>
<bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="spring-active-queue"/>
</bean>
<!-- Spring的JMS模版工具 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="jmsFactory" />
<property name="defaultDestination" ref="destinationQueue"/>
<property name="messageConverter">
<bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
</property>
</bean>
</beans>
4.3.队列
生产者
@Service
public class SpringMQ_Producer {
@Autowired
private JmsTemplate jmsTemplate;
public static void main(String[] args) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
SpringMQ_Producer producer = (SpringMQ_Producer) applicationContext.getBean("springMQ_Producer");
producer.jmsTemplate.send(new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage("Spring+Active。。。。");
return textMessage;
}
});
}
}
消费者
@Service
public class SpringMQ_Consumer {
@Autowired
private JmsTemplate jmsTemplate;
public static void main(String[] args) throws JMSException {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
SpringMQ_Consumer consumer = (SpringMQ_Consumer) applicationContext.getBean("springMQ_Consumer");
// TextMessage receive =(TextMessage) consumer.jmsTemplate.receive();
String receive =(String) consumer.jmsTemplate.receiveAndConvert();
System.out.println("消费者："+receive);
}
}
4.4 主题
applicationContext.xml
Spring配置文件，新增主题topic
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="com.atguigu.activemq"></context:component-scan>
<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://10.112.70.211:61616"/>
</bean>
</property>
<property name="maxConnections" value="100"></property>
</bean>
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-active-topic"/>
</bean>
<!-- Spring的JMS模版工具 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="jmsFactory" />
<property name="defaultDestination" ref="destinationTopic"/>
<property name="messageConverter">
<bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
</property>
</bean>
</beans>
生产者
消费者 （同队列）
4.5.在spring中里边实现消费者不启动，直接通过配置监听完成
applicationContext.xml
<bean id="queueListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsFactory" />
<property name="destination" ref="destinationTopic" />
<property name="messageListener" ref="myMessageListener" />
</bean>
监听类
@Service
public class MyMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
if (null != message && message instanceof TextMessage){
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("spring消费者接收到消息："+textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
5.ActiveMQ传输协议
5.1TCP默认的
5.2 NIO
----------------------总结
5.3 修改为NIO
Main.java
public static final String URL = "nio://10.112.70.211:61616";
public static final String queue_name = "Transport";
6.ActiveMQ消息持久化与存储
6.1持久化
6.2 消息存储
6.2.1KahaDB 消息存储（默认）
6.2.2 JDBC 消息存储
消息基于JDBC存储的！
a.MQ+Mysql
b.添加mysql数据库的驱动包到lib文件夹
c.jdbcPersistanceAdapter配置
d.数据库连接池配置
<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://10.112.70.211:3306/activemq?relaxAutoCommit=true"/>
<property name="username" value="root"/>
<property name="password" value="123456"/>
<property name="poolPreparedStatements" value="true"/>
</bean>
e.建库建表说明
-------建一个名为activemq的数据库如果新建数据库OK+上述配置OK+代码运行OK，三张表会自动建立!
表activemq_msgs:
表activemq_acks：
表activemq_lock：
f.代码运行验证producer.setDeliveryMode(DeliveryMode.PERSISTENT);开启持久化
队列、订阅生产者和消费者的java代码
g.数据库情况mysql
queue:在运行消费看看表msgs数据的变化情况
topic:先启动消费者订阅在运行生产者，看acks
-----------------总结---------------------
6.2.3 ActiveMQ Message Store (了解)
基于文件的存储方式，是以前的默认消息存储，现在不用了
--------------坑----------------------------
上一篇： LeetCode 71. Simplify Path
下一篇： Tomcat初认识