yangyoupeng / microservice

Go to a project
Toggle navigation Toggle navigation pinning
yangyoupeng
5d283081 1 parent 93c6cf5e

新增CloudPullTopicConsumer已经测试例，作用是检查队列是否存在、若不存在就创建

Inline Side-by-side
Showing 6 changed files with 318 additions and 7 deletions
1 +package com.zhaoonline.message.queue;
2 +
3 +import java.util.Vector;
4 +
5 +import org.apache.commons.pool.impl.GenericObjectPool;
6 +import org.slf4j.Logger;
7 +import org.slf4j.LoggerFactory;
8 +
9 +import com.aliyun.mns.client.CloudPullTopic;
10 +import com.aliyun.mns.client.CloudQueue;
11 +import com.aliyun.mns.client.MNSClient;
12 +import com.aliyun.mns.model.Message;
13 +import com.aliyun.mns.model.QueueMeta;
14 +import com.aliyun.mns.model.TopicMeta;
15 +
16 +public class CloudPullTopicConsumer {
17 +
18 + private static Logger logger=LoggerFactory.getLogger(CloudPullTopicConsumer.class);
19 +
20 + private TopicMeta topicMeta =new TopicMeta();
21 + private String queueName;
22 + private GenericObjectPool<MNSClient> pool;
23 + private QueueMeta queueMetaTemplate=new QueueMeta();
24 + private CloudQueue queue=null;
25 + private MNSClient mnsClient =null;
26 + private boolean initiated=false;
27 + public CloudPullTopicConsumer(String topic,GenericObjectPool<MNSClient> pool,String queue) throws Exception {
28 + this(topic,pool,null,queue);
29 + }
30 +
31 + public CloudPullTopicConsumer(String topic,GenericObjectPool<MNSClient> pool,QueueMeta queueMetaTemplate,String queue) throws Exception {
32 + this(topic,null,pool,queueMetaTemplate,queue);
33 + }
34 +
35 + public CloudPullTopicConsumer(String topic,TopicMeta topicMeta,GenericObjectPool<MNSClient> pool,QueueMeta queueMetaTemplate,String queue) throws Exception {
36 + this(topic,topicMeta,pool,queueMetaTemplate,queue,true);
37 + }
38 +
39 + public CloudPullTopicConsumer(String topic,TopicMeta topicMeta,GenericObjectPool<MNSClient> pool,QueueMeta queueMetaTemplate,String queueName,boolean initiate) throws Exception {
40 + if(topicMeta != null){
41 + this.topicMeta =topicMeta;
42 + }
43 + this.topicMeta.setTopicName(topic);
44 + this.queueName = queueName;
45 + if(queueMetaTemplate != null){
46 + this.queueMetaTemplate=queueMetaTemplate;
47 + }else{
48 + this.queueMetaTemplate.setPollingWaitSeconds(30);
49 + }
50 + initiated=initiate;
51 + this.pool=pool;
52 + init();
53 + }
54 +
55 + public void init() throws Exception{
56 + mnsClient=pool.borrowObject();
57 + queue=mnsClient.getQueueRef(this.queueName);
58 + if(initiated){
59 + if(queueNotExisted(queue)){
60 + logger.info("queue:[{}] is not existed going to create it with queueMeta:[{}].",queue,queueMetaTemplate);
61 + subScribeToTopic(queue);
62 + }
63 + }
64 + }
65 +
66 + public boolean queueExisted(){
67 + return queue.isQueueExist();
68 + }
69 + private void subScribeToTopic(CloudQueue queue) throws Exception {
70 + Vector<String> queueNameVector=new Vector<>(1);
71 + queueNameVector.add(queueName);
72 + CloudPullTopic topic = mnsClient.createPullTopic(topicMeta, queueNameVector, true, queueMetaTemplate);
73 + logger.info("create a new queue:[{}] with queueMeta for topic:[{}] with topicMeta:[{}]",queue,queueMetaTemplate,topic,topicMeta);
74 + }
75 +
76 +
77 + private boolean queueNotExisted(CloudQueue queue) {
78 + return !queue.isQueueExist();
79 + }
80 +
81 + public Message popMessage(){
82 + return queue.popMessage();
83 + }
84 +
85 + public MessageObject popMessageObject(){
86 + Message result=queue.popMessage();
87 + return MessageObject.wrapMessage(result);
88 + }
89 + public Message popMessage(int timeount){
90 + return queue.popMessage(timeount);
91 + }
92 +
93 + public MessageObject popMessageObject(int timeount){
94 + Message result=queue.popMessage(timeount);
95 + return MessageObject.wrapMessage(result);
96 + }
97 +
98 +
99 + public Message peekMessage() {
100 + return queue.peekMessage();
101 + }
102 +
103 + public MessageObject peekMessageObject() {
104 + Message result=queue.peekMessage();
105 + return MessageObject.wrapMessage(result);
106 + }
107 +
108 +
109 + public void deleteAlreadyGetMessage(Message result) {
110 + if(result !=null){
111 + queue.deleteMessage(result.getReceiptHandle());
112 + }
113 + }
114 +
115 +
116 + public void close(){
117 + if(mnsClient!=null){
118 + try {
119 + pool.returnObject(mnsClient);
120 + } catch (Exception e) {
121 + e.printStackTrace();
122 + }
123 + }
124 +
125 + }
126 +
127 +
128 +}
...@@ -145,6 +145,22 @@ public class CloudPullTopicProducer implements ProducerInt<MessageObject,TopicMe ...@@ -145,6 +145,22 @@ public class CloudPullTopicProducer implements ProducerInt<MessageObject,TopicMe
145 return cloudPullTopic.getQueueNameList(); 145 return cloudPullTopic.getQueueNameList();
146 } 146 }
147 147
148 + /**
149 + * Method name: deleteTopic <BR>
150 + * Description: 该delete方法的作用是需要做重点说明：
151 + * 调用deleteTopic(true) ,要想连同topic和queue一起删除，需要在构建CloudPullTopicProducer的使用
152 + * {@code
153 + * boolean createQueue=true;
154 + * CloudPullTopicProducer cloudPullTopicProducer = CloudTopicProducerBuilder.buildCloudTopic(topicName,createQueue, 队列名1,队列名2)
155 + * 或者使用MNSClient来创建：
156 + * boolean createQueue=true;
157 + * CloudPullTopic cloudPullTopic = mnsClient.createPullTopic(topicMeta, queueNameVector,createQueue,queueMetaTemplate);
158 + * }
159 + *直接只能删除初始化时输入的队列名。如上面的["队列名1,队列名2"] 或者是queueNameVector
160 + * <BR>
161 + * Remark: <BR>
162 + * @param needDeleteQueues void<BR>
163 + */
148 public void deleteTopic(boolean needDeleteQueues){ 164 public void deleteTopic(boolean needDeleteQueues){
149 cloudPullTopic.delete(needDeleteQueues); 165 cloudPullTopic.delete(needDeleteQueues);
150 } 166 }
......
...@@ -64,6 +64,48 @@ public class CloudTopicProducerBuilder { ...@@ -64,6 +64,48 @@ public class CloudTopicProducerBuilder {
64 } 64 }
65 65
66 /** 66 /**
67 + * Method name: buildTopicPullConsumer <BR>
68 + * Description: topicPullConsumer在检查queue是否存在，不存在就会创建.topicMeta和queue都采用默认值 <BR>
69 + * Remark: <BR>
70 + * @param queue
71 + * @return Consumer<BR>
72 + * @throws Exception
73 + */
74 + public CloudPullTopicConsumer buildTopicPullConsumer(String topic,String queue) throws Exception{
75 + CloudPullTopicConsumer consumer=new CloudPullTopicConsumer(topic,pool,queue);
76 + return consumer;
77 + }
78 +
79 + /**
80 + * Method name: buildTopicPullConsumer <BR>
81 + * Description: <BR>
82 + * Remark: <BR>
83 + * @param topicName
84 + * @param queue
85 + * @param initiate 是否需要立刻init，如果为false，那么需要手动调用{@link CloudPullTopicConsumer#init()}方法,主要供测试用
86 + * @return
87 + * @throws Exception CloudPullTopicConsumer<BR>
88 + */
89 + public CloudPullTopicConsumer buildTopicPullConsumer(String topicName, String queue, boolean initiate) throws Exception {
90 + CloudPullTopicConsumer consumer=new CloudPullTopicConsumer(topicName,null,pool,null,queue,initiate);
91 + return consumer;
92 + }
93 + /**
94 + * Method name: buildTopicPullConsumer <BR>
95 + * Description: topicPullConsumer在检查queue是否存在，不存在就会创建。指定使用topMeta以及queueMeta<BR>
96 + * Remark: <BR>
97 + * @param queue
98 + * @return Consumer<BR>
99 + * @throws Exception
100 + */
101 + public CloudPullTopicConsumer buildTopicPullConsumer(String topic,TopicMeta topicMeta,QueueMeta queueMeta,String queue) throws Exception{
102 + CloudPullTopicConsumer consumer=new CloudPullTopicConsumer(topic,topicMeta, pool, queueMeta,queue);
103 + return consumer;
104 + }
105 +
106 +
107 +
108 + /**
67 * Method name: buildConsumer <BR> 109 * Method name: buildConsumer <BR>
68 * Description: 构建queue的consumer <BR> 110 * Description: 构建queue的consumer <BR>
69 * Remark: <BR> 111 * Remark: <BR>
...@@ -129,4 +171,6 @@ public class CloudTopicProducerBuilder { ...@@ -129,4 +171,6 @@ public class CloudTopicProducerBuilder {
129 logger.error("error when pool closed,cause by ",e.getMessage()); 171 logger.error("error when pool closed,cause by ",e.getMessage());
130 } 172 }
131 } 173 }
174 +
175 +
132 } 176 }
......
...@@ -70,21 +70,34 @@ public class Consumer { ...@@ -70,21 +70,34 @@ public class Consumer {
70 * @param timeount 70 * @param timeount
71 * @return Message<BR> 71 * @return Message<BR>
72 */ 72 */
73 - public MessageObject popMessage(int timeount){ 73 + public Message popMessage(int timeount){
74 + return queueForConsumer.popMessage(timeount);
75 + }
76 +
77 +
78 + public MessageObject popMessageObject(int timeount){
74 Message result=queueForConsumer.popMessage(timeount); 79 Message result=queueForConsumer.popMessage(timeount);
75 return MessageObject.wrapMessage(result); 80 return MessageObject.wrapMessage(result);
76 } 81 }
77 82
83 +
84 +
78 /** 85 /**
79 * Method name: popMessage <BR> 86 * Method name: popMessage <BR>
80 * Description: pop完了之后会自动删除message。 <BR> 87 * Description: pop完了之后会自动删除message。 <BR>
81 * Remark: <BR> 88 * Remark: <BR>
82 * @return MessageObject<BR> 89 * @return MessageObject<BR>
83 */ 90 */
84 - public MessageObject popMessage(){ 91 + public Message popMessage(){
92 + return queueForConsumer.popMessage();
93 + }
94 +
95 + public MessageObject popMessageObject(){
85 Message result=queueForConsumer.popMessage(); 96 Message result=queueForConsumer.popMessage();
86 return MessageObject.wrapMessage(result); 97 return MessageObject.wrapMessage(result);
87 } 98 }
99 +
100 +
88 // /** 101 // /**
89 // * Method name: popMessageAndDelete <BR> 102 // * Method name: popMessageAndDelete <BR>
90 // * Description: 获取消息之后，删除队列中的消息，此时才能继续消费下一条消息，若队列中暂时没有消息，则等待timeout时间 <BR> 103 // * Description: 获取消息之后，删除队列中的消息，此时才能继续消费下一条消息，若队列中暂时没有消息，则等待timeout时间 <BR>
...@@ -143,12 +156,14 @@ public class Consumer { ...@@ -143,12 +156,14 @@ public class Consumer {
143 * Remark: <BR> 156 * Remark: <BR>
144 * @return MessageObject<BR> 157 * @return MessageObject<BR>
145 */ 158 */
146 - public MessageObject peekMessage() { 159 + public MessageObject peekMessageObject() {
147 Message result=queueForConsumer.peekMessage(); 160 Message result=queueForConsumer.peekMessage();
148 return MessageObject.wrapMessage(result); 161 return MessageObject.wrapMessage(result);
149 } 162 }
150 163
151 - 164 + public Message peekMessage() {
165 + return queueForConsumer.peekMessage();
166 + }
152 167
153 168
154 } 169 }
......
1 +package com.zhaoonline.message.queue;
2 +
3 +import org.junit.Assert;
4 +import org.junit.Before;
5 +import org.junit.Rule;
6 +import org.junit.Test;
7 +import org.junit.rules.ExpectedException;
8 +
9 +import com.aliyun.mns.model.Message;
10 +
11 +
12 +public class TestCloudPullTopicConsumer {
13 +
14 + CloudTopicProducerBuilder builder=null;
15 + AliMNSConfig config=null;
16 +
17 + @Rule
18 + public ExpectedException thrown = ExpectedException.none();
19 + @Before
20 + public void prepare(){
21 + config=new AliMNSConfig();
22 + config.setAccountEndpoint("http://1384553363882565.mns.cn-hangzhou.aliyuncs.com");
23 + config.setAccessId("QfXG4pxBPbUEVk0p");
24 + config.setAccessKey("SI56mM6jEHE0pRzb7oawKeX8CiXC83");
25 + builder=new CloudTopicProducerBuilder(config);
26 +
27 + }
28 +
29 +
30 + /**
31 + * Method name: testCheckQueueIsExisted <BR>
32 + * Description: 测试queue是否已经存在 <BR>
33 + * Remark: <BR>
34 + * @throws Exception void<BR>
35 + */
36 + @Test
37 + public void testCheckQueueIsExisted() throws Exception{
38 + String topicName="testTopic1";
39 + //createQueue=false的是时候，只会创建topic，而不会创建queue，测试如果publish的时候就会报异常错误
40 + String testQueue1 = "testQu1eue";
41 + String testQueue2 = "testQ1ueu";
42 + TopicProducer topic=builder.buildTopic(topicName);
43 +
44 + CloudPullTopicProducer topicProducer=builder.buildCloudTopic(topicName,true, testQueue1,testQueue2);
45 + topicProducer.deleteTopic(true);
46 +
47 +
48 + boolean initiate=false;
49 + CloudPullTopicConsumer consumer=builder.buildTopicPullConsumer(topicName, testQueue1,initiate);
50 +
51 + Assert.assertFalse(consumer.queueExisted());
52 +
53 + topicProducer.deleteTopic(true);
54 +
55 + //
56 + CloudPullTopicConsumer consumer2=builder.buildTopicPullConsumer(topicName, testQueue1,true);
57 +
58 + Assert.assertTrue(consumer2.queueExisted());
59 +
60 + topicProducer.deleteTopic(true);
61 + }
62 +
63 +
64 + /**
65 + * Method name: testAddNewQueue2ExitedTopic <BR>
66 + * Description: 新增一个队列到已经存在的topic上面去 <BR>
67 + * Remark: <BR>
68 + * @throws Exception void<BR>
69 + */
70 + @Test
71 + public void testAddNewQueue2ExitedTopic() throws Exception{
72 + String topicName="testTopic1";
73 + //createQueue=false的是时候，只会创建topic，而不会创建queue，测试如果publish的时候就会报异常错误
74 + String testQueue1 = "testQu1eue";
75 + String testQueue2 = "testQ1ueu";
76 + TopicProducer topic=builder.buildTopic(topicName);
77 + CloudPullTopicProducer topicProducer=builder.buildCloudTopic(topicName,true, testQueue1,testQueue2);
78 + topicProducer.deleteTopic(true);
79 +
80 + //创建一个topic,并且他包含一个testQueue1;
81 + CloudPullTopicProducer topicProducer2=builder.buildCloudTopic(topicName,true, testQueue1);
82 +
83 +
84 + topicProducer2.send("hellow world1");
85 +
86 + CloudPullTopicConsumer consumer1=builder.buildTopicPullConsumer(topicName, testQueue1);
87 + Message msg1=consumer1.popMessage();
88 + Assert.assertEquals("hellow world1", msg1.getMessageBodyAsRawString());
89 +
90 +
91 + //新建一个consumer。他消费testQueue2，这个testQueue2并不存在
92 + CloudPullTopicConsumer consumer2=builder.buildTopicPullConsumer(topicName, testQueue2);
93 +
94 +
95 + topicProducer2.send("hellow world2");
96 +
97 + Message msg2_1=consumer1.popMessage();
98 + Message msg2_2=consumer2.popMessage();
99 +
100 + Assert.assertEquals("hellow world2", msg2_1.getMessageBodyAsRawString());
101 +
102 + Assert.assertEquals("hellow world2", msg2_2.getMessageBodyAsRawString());
103 +
104 + topicProducer.deleteTopic(true);
105 + }
106 +
107 +
108 +}
...@@ -42,14 +42,14 @@ public class TestP2PProducer { ...@@ -42,14 +42,14 @@ public class TestP2PProducer {
42 Message message=producer.send("helloworld"); 42 Message message=producer.send("helloworld");
43 System.out.println(message.getReceiptHandle()); 43 System.out.println(message.getReceiptHandle());
44 Consumer consumer=builder.buildConsumer("p2pTestQueue"); 44 Consumer consumer=builder.buildConsumer("p2pTestQueue");
45 - MessageObject object= consumer.popMessage(); 45 + MessageObject object= consumer.popMessageObject();
46 Assert.assertNotNull(object.getMessage()); 46 Assert.assertNotNull(object.getMessage());
47 System.out.println(object.getMessage().getReceiptHandle()); 47 System.out.println(object.getMessage().getReceiptHandle());
48 System.out.println("----first pop-----"+object.getMessage()); 48 System.out.println("----first pop-----"+object.getMessage());
49 System.out.println("----first pop-----"+object.getMessage().getMessageBodyAsString()); 49 System.out.println("----first pop-----"+object.getMessage().getMessageBodyAsString());
50 50
51 51
52 - MessageObject object2= consumer.popMessage(); 52 + MessageObject object2= consumer.popMessageObject();
53 Assert.assertNull(object2.getMessage()); 53 Assert.assertNull(object2.getMessage());
54 System.out.println("--second pop should null--"+object2.getMessage()); 54 System.out.println("--second pop should null--"+object2.getMessage());
55 55
...@@ -68,7 +68,7 @@ public class TestP2PProducer { ...@@ -68,7 +68,7 @@ public class TestP2PProducer {
68 System.out.println(message.getReceiptHandle()); 68 System.out.println(message.getReceiptHandle());
69 69
70 Consumer consumer=builder.buildConsumer("p2pTestQueue"); 70 Consumer consumer=builder.buildConsumer("p2pTestQueue");
71 - MessageObject object= consumer.peekMessage(); 71 + MessageObject object= consumer.peekMessageObject();
72 72
73 System.out.println(object.getMessage().getReceiptHandle()); 73 System.out.println(object.getMessage().getReceiptHandle());
74 74
......