Skip to content

Commit a0038e6

Browse files
KevinBlandyKevinBlandy
committed
首次提交
0 parents  commit a0038e6

1,480 files changed

Lines changed: 110971 additions & 0 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
/private/

‎ActiveMQ/ActiveMQ-API.java‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
Connection
2+
* JMS定义的接口,代表跟MQ服务器的一个连接,通常一个服务器只用定义一个接口
3+
* 当一个 Connection 定义的时候,默认是关闭状态.需要手动开启
4+
* 一个 Connection 可以创建N多个 Session
5+
* 当程序执行完毕,应该关闭 Connection,它会自动的去关闭 Session,MessageProducer 和 MessageConsumer
6+
7+
8+
Session
9+
* JMS定义的接口,由 Connection 生成,一个或者多个. Session 是一个收发消息的线程
10+
* 可以使用 Session 创建 MessageProducer,MessageConsumer和Message
11+
* Session 可以被事务化,或者非事务化,通过向创建 Session 的方法,传递 boolean 参数控制
12+
Session session = connection.createSession(boolean transacted, int acknowledgeMode);
13+
transacted:是否创建事务session
14+
'本地事务'
15+
结束事务有两种方法,提交或者回滚
16+
当一个事务提交,消息被处理,如果事务中有一个步骤失败,事务就回滚.这个事务中已执行的动作将会被撤销.
17+
在发送消息最后也必须要用 session.commit();方法表示提交事务
18+
'所谓的事务回滚,其实就是---不发送消息到message,在回滚/提交之前.你可以进行大量的消息写入队列或者其他操作,提交就去了服务器,回滚就啥也没了'
19+
20+
acknowledgeMode:消息签收模式
21+
Session.AUTO_ACKNOWLEDGE
22+
当消费者从receive或者onMessage成功返回时,Session自动签收消费者这条消息的收条
23+
'自动的签收消息'
24+
25+
Session.CLIENT_ACKNOWLEDGE
26+
消费者通过调用消息(Message)的acknowledge方法签收消息
27+
这种情况下,签收发生在 Session 层面
28+
签收一个已经消费的消息,会自动的签收这个 Session 所有已经消息的收条
29+
'手动的签收消息'
30+
31+
Session.DUPS_OK_ACKNOWLEDGE
32+
这参数,Session不会确保对传送消息的签收,它可能引起消息的重复,但是降低了Session的开销
33+
只有消费者允许重复的消息,才使用
34+
'不执行签收,N个消费者可能会消费同一条消息,或者一个消费者重复消费同一条消息'
35+
36+
37+
MessageProducer
38+
* 是一个由 Session 创建的对象,用来向 Destination 发送消息
39+
* session创建
40+
MessageProducer createProducer(Destination destination)
41+
* 发送消息
42+
void send(Message message)
43+
void send(Message message, int deliveryMode, int priority, long timeToLive)
44+
void send(Destination destination,Message message)
45+
void send(Destination destination,Message message,int deliveryMode,int priority,long timeToLive)
46+
* 参数介绍
47+
deliveryMode
48+
* 消息传送模式(是否持久化)
49+
DeliveryMode.NON_PERSISTENT //不持久化,性能好.数据易丢失
50+
DeliveryMode.PERSISTENT(默认) //持久化,性能差.数据不易丢失
51+
52+
priority
53+
* 消息优先级
54+
优先级从 0-9 一共十个级别(默认值=4),JMS不要求严格按照10个优先级发送消息
55+
0-4 普通消息 //
56+
5-9 加急消息 //该消息,理论会比普通消息先到达.注意哟,是理论
57+
58+
timeToLive
59+
* 过期时间
60+
默认清空下,消息永远不会过期,如果需要消息在一定时间内失去意义,那么可以设置该选项
61+
单位是毫秒
62+
63+
64+
MessageConsumer
65+
* 是一个由 Session 创建的对象,用来从 Destination 接收消息
66+
* session创建
67+
MessageConsumer createConsumer(Destination destination);
68+
MessageConsumer createConsumer(Destination destination,String messageSelector);
69+
MessageConsumer createConsumer(Destination destination,String messageSelector,boolean NoLocal);
70+
* 参数介绍
71+
messageSelector
72+
* 消息选择器(过滤器)
73+
* 是基于 SQL92的语法,其实就是某种规则的字符串(例如:正则)
74+
75+
NoLocal
76+
* 默认为 false
77+
* 为 true 的话,只能接收和自己相同连接(Connection)所发布的消息,此标识仅仅'适用于主题,不适用于队列'
78+
name
79+
* 明确,订阅主题的时候,声明的订阅名称.持久订阅时需要此参数
80+
81+
82+
TopicSubscriber createDurableSubscriber(Topic topic,String name);
83+
TopicSubscriber createDurableSubscriber(Topic topic,String name,String messageSelector,boolean noLocal)
84+

‎ActiveMQ/ActiveMQ-Destination高级特性.java‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
-------------------------------------
2+
ActiveMQ-Destination高级特性 |
3+
-------------------------------------
4+
5+
6+
-------------------------------------
7+
ActiveMQ-Wildcards |
8+
-------------------------------------
9+
# Wildcards 通配符
10+
# 这东西并不是JMS的规范,而是ActiveMQ的扩展
11+
# ActiveMQ支持一下三种:wildcards
12+
①:".",用于作为路径上名字间的分隔符
13+
②:"*",用于匹配任何名字
14+
③:">",用于递归匹配任何以这个名字开始的 Destination
15+
16+
17+
-------------------------------------
18+
ActiveMQ-Composite Destinations |
19+
-------------------------------------
20+
# 组合队列
21+
# 组合队列,允许用一个虚拟的 Destination 代表多个 Destination
22+
# 说白了就是,同时往多个队列,或者 topic 推送消息
23+
# 代码
24+
/**
25+
此处就是创建了带有三个队列的一个 Queue
26+
多个Queue用逗号隔开
27+
*/
28+
Queue queue = new ActiveMQQueue("queue#1,queue#2,queue#3");
29+
/**
30+
通过这个,创建消息生产者
31+
*/
32+
MessageProducer producer = session.createProducer(queue);
33+
34+
# 也可以混合发送,就是一个 Queue 里面不进有队列,还有 Topic
35+
# 代码
36+
Queue queue = new ActiveMQQueue("queue#1,topic://topic#1");
37+
MessageProducer producer = session.createProducer(queue);
38+
* 跟上面其实一个德行,无非就是.话题的格式是用: topic://
39+
40+
# 也可以通过配置来完成...不过很傻逼,不学
41+
42+
43+
-------------------------------------
44+
ActiveMQ-Delete Inactive Destination |
45+
-------------------------------------
46+
# 自动删除队列
47+
# 一般清空下,ActiveMQ,的 Queue,在不使用之后,可以通过WEB控制,或者JMX的方式来进行删除
48+
# 也可以通过配置,让MQ自动探测到无用的队列(一定时间内为空的队列),进行删除.释放资源
49+
50+
<broker schedulePeriodForDestinationPurge="10000">
51+
<destinationPolicy>
52+
<policyMap>
53+
<policyEntries>
54+
<policyEntry queue=">" gcInactiveDestinations="true" inactiveTimoutBeforeGC="30000"/>
55+
</policyEntries>
56+
</policyMap>
57+
</destinationPolicy>
58+
</broker>
59+
60+
* schedulePeriodForDestinationPurge :多长时间检查一次,这里设置为10S,默认为0
61+
* inactiveTimeoutBeforGC :当 Destination 为空后,多长时间被删除,这里是30s,默认为60
62+
gcInactiveDestinations :设置删除掉不活动的队列,默认为 false
63+
64+
-------------------------------------
65+
ActiveMQ- Destination Options |
66+
-------------------------------------
67+
# 队列选项,这东西也并不是JMS规范中的东西
68+
# 其实就是说可以在队列名称后面向HTTP的URL传参一样传递一些参数
69+
# 参数有
70+
consumer.prefetchSize
71+
* consumer持久的未ack的最大消息数量,默认值为 variable
72+
consumer.maximumPendingMessageLimit
73+
* 用来控制非持久化的topic在存在慢消费者的情况下丢弃的数量
74+
consumer.noLocal
75+
* 默认为 false
76+
consumer.dispatchAsync
77+
* 是否异步分发,默认为 true
78+
consumer.retroactive
79+
* 是否回溯消费者,默认为 false
80+
consumer.selector
81+
* JMS的selector,默认为 null
82+
consumer.exclusive
83+
* 是否为独占消费者,默认为 false
84+
consumer.priority
85+
* 设置消费者的优先级别,默认 0
86+
87+
# Demo
88+
Queue queue = new ActiveMQQueue("queue#1?consumer.exclusive=false&consumer.priority=4");
89+
MessageConsumer messageConsumer = session.createConsumer(queue);

‎ActiveMQ/ActiveMQ-HelloWorld.java‎

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
package com.kevin.demo.activemq.helloworld;
2+
import javax.jms.Connection;
3+
import javax.jms.ConnectionFactory;
4+
import javax.jms.DeliveryMode;
5+
import javax.jms.Destination;
6+
import javax.jms.JMSException;
7+
import javax.jms.MessageProducer;
8+
import javax.jms.Session;
9+
import javax.jms.TextMessage;
10+
import org.apache.activemq.ActiveMQConnectionFactory;
11+
/**
12+
* 消息生产者
13+
* */
14+
public class Provider {
15+
public static void main(String[] args) throws JMSException {
16+
/**
17+
* 1,创建连接工场.使用JMS规范
18+
* 用户名和密码使用默认,连接方式为TCP
19+
* */
20+
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
21+
ActiveMQConnectionFactory.DEFAULT_USER,
22+
ActiveMQConnectionFactory.DEFAULT_PASSWORD,
23+
"tcp://123.207.122.145:61616");
24+
/***
25+
* 2,通过工厂获取一个连接,并且打开它(在创建连接的时候,也是可以设置用户名和密码)
26+
* 默认是关闭状态
27+
* */
28+
Connection connection = connectionFactory.createConnection();
29+
connection.start();
30+
/**
31+
* 3,通过连接,获取会话
32+
* 参数1,是否启用事务(不启用)
33+
* 参数2,消息签收模式(消费者手动签收)
34+
* */
35+
Session session = connection.createSession(Boolean.TRUE,Session.CLIENT_ACKNOWLEDGE);
36+
/**
37+
* 4,通过会话创建 Destination 对象
38+
* Destination ,用来描述 消费者 和 消息来源的对象.其实是一个高层的抽象
39+
* 在PTP模式中,Destination 被称为:Queue,也就是队列
40+
* 在Pub/Sub模式中,Destination 被称为:Topic,也就是主题
41+
* */
42+
Destination destination = session.createQueue("queue#1"); //PTP模式,创建的是队列,并且指定队列名称
43+
/**
44+
* 5,通过会话创建 消息生产者 MessageProducer
45+
* 构造传入 Destination 描述对象
46+
* 这里的构造,也可以指定为 null,在 messageProducer 执行 send 发送消息的时候,才指定 Destination
47+
* */
48+
MessageProducer messageProducer = session.createProducer(destination);
49+
/**
50+
* 6,设置 消息生产者的一些属性
51+
* 也可以在执行 send 的时候,去指定属性
52+
* */
53+
messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); //设置为非持久化
54+
messageProducer.setPriority(9); //设置消息优先级别
55+
/**
56+
* 7,最后,通过会话,使用JMS规范中的 TextMessage 形式创建数据,并使用 消息生产者MessageProducer 来发送消息
57+
* */
58+
for(int x = 0;x < 5;x++){
59+
TextMessage message = session.createTextMessage("消息签收模式-客户端签收消息#" + x); //直接指定消息内容
60+
//message.setText("hello"); 也可以创建后添加消息
61+
messageProducer.send(message);
62+
//N多构造,可以在发送的时候,配置一些属性
63+
//messageProducer.send(destination, message, DeliveryMode.NON_PERSISTENT, Priority.DEBUG_INT, 9999999);
64+
}
65+
session.commit();
66+
/**
67+
* 8,关闭资源
68+
* */
69+
if(connection != null){
70+
/**
71+
* 仅需要关闭此连接,它会自己去关闭其他的有关连接
72+
* */
73+
connection.close();
74+
}
75+
}
76+
}
77+
78+
79+
80+
-----------------------------------------------------------------------------------------------
81+
package com.kevin.demo.activemq.helloworld;
82+
import javax.jms.Connection;
83+
import javax.jms.ConnectionFactory;
84+
import javax.jms.Destination;
85+
import javax.jms.JMSException;
86+
import javax.jms.MessageConsumer;
87+
import javax.jms.Session;
88+
import javax.jms.TextMessage;
89+
import org.apache.activemq.ActiveMQConnectionFactory;
90+
/**
91+
* 消息消费者
92+
* */
93+
public class Consumer {
94+
public static void main(String[] args) throws JMSException {
95+
/**
96+
* 1,建立ConnectionFactory工厂
97+
* 使用默认的用户名和密码,连接方式为TCP
98+
* */
99+
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
100+
ActiveMQConnectionFactory.DEFAULT_USER,
101+
ActiveMQConnectionFactory.DEFAULT_PASSWORD,
102+
"tcp://123.207.122.145:61616");
103+
/**
104+
* 2,通过ConnectionFactory工厂,获取到连接 Connection,并且打开连接(默认为关闭状态)
105+
* */
106+
Connection connection = connectionFactory.createConnection();
107+
connection.start();
108+
/**
109+
* 3,通过连接,创建会话,Session
110+
* 不启用事务
111+
* 消息签收模式为 消费者手动签收
112+
* */
113+
Session session = connection.createSession(Boolean.FALSE,Session.CLIENT_ACKNOWLEDGE);
114+
/**
115+
* 4,通过 会话 创建 Destination 对象
116+
* Destination ,用来描述 消费者 和 消息来源的对象.其实是一个高层的抽象
117+
* 在PTP模式中,Destination 被称为:Queue,也就是队列
118+
* 在Pub/Sub模式中,Destination 被称为:Topic,也就是主题
119+
* */
120+
Destination destination = session.createQueue("queue#1"); //指定队列的名称
121+
/**
122+
* 5,通过Session创建消息接收对象
123+
* 构造传入 Destination 描述对象
124+
* */
125+
MessageConsumer messageConsumer = session.createConsumer(destination);
126+
/**
127+
* 6,使用接收对象(MessageConsumer),的 receive 来创建 ,JMS规范的 TextMessage 对象,
128+
* */
129+
while(true){
130+
/**
131+
* receive(),这个方法有很多重载形式
132+
* 1,空参,阻塞,不多解释.BIO那种节奏
133+
* 2,Long,阻塞时间,如果超时.就会往下执行
134+
* 3,NoWait(),程序执行到这里检查是否有数据,有数据就消费.不然直接往下执行.不会等待,更不会阻塞
135+
* */
136+
TextMessage message = (TextMessage) messageConsumer.receive();
137+
if(message == null){
138+
break;
139+
}
140+
message.acknowledge(); //手动签收消息,其实就是另起线程.TCP通知MQ,这个消息我成功消费了
141+
System.out.println(message.toString());
142+
System.err.println("收到消息:" + message.getText());
143+
}
144+
/**
145+
*7, 关闭资源
146+
* */
147+
if(connection != null){
148+
connection.close();
149+
}
150+
}
151+
}

0 commit comments

Comments
 (0)