博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
【Active入门-2】ActiveMQ学习-生产者与消费者
阅读量:5128 次
发布时间:2019-06-13

本文共 4698 字,大约阅读时间需要 15 分钟。

1个生产者,1个消费者,使用Queue:

方式1:
生产者将消息发送到Queue中,退出;
然后运行消费者:
可以看到,可以接收到消息。

方式2:
先运行消费者程序:
然后运行生产者:
消费者见下图:

1个生产者,2个消费者,使用Queue

先运行消费者1:
在运行消费者2:
接下来运行生产者:
下面是消费者消费情况:
总结:
1. 使用Queue时,生产者只要将Message发送到MQ服务器端,消费者就可以进行消费,而无需生产者程序一直运行;
2. 消息是按照先入先出的顺序,一旦有消费者将Message消费,该Message就会从MQ服务器队列中删去;
3. 有文章说,“生产者”<-->"消费者"是一对一的关系,其实并不准确,从上面可以看出,一个生产者产生的消息,可以被
多个消费者进行消费,只不过多个消费者在消费消息时是竞争的关系,先得到的先消费,一旦消费完成,该消息就会出队列,
就不能被其他消费者再消费了,即“一次性消费”。
特点:
1.“离线一次性消费”;
离线:指生产者在发送消息时,不需要消费者在线,生产者只需要将消息发送到MQ队列中,消费者可以稍后上线取消息;
一次性消费:指消费者之间存在竞争关系,任何一个消费者将消息消费掉之后,其他消费者都不能再进行消费;
若是只有1个生产者和1个消费者,就是我们熟悉的“点对点”通信了;
类似场景对比:
类似送快递,快递员(producer)将快递(Message)放到指定地点(destination)后,就可以离开了,
拿快递的人(customer)在接收到通知后,到指定地点(destination)去取快递(Message)就可以了。
当然,取快递时可能要进行身份验证,这就涉及到创建连接(connection)时,需要指定用户名和密码了。
还有就是,实际生活中,当快递员把快递放好之后,照理说应该通知客户去哪里取快递,而ActiveMq帮我们
做好了一切,通知的工作Activemq会帮我们实现,而无需我们亲自编码通知消费者,生产者只需要将Message
放到Mq中即可,通知消费者的工作,mq会帮我们处理。

消费者程序:

 
  1. package com.ll.activemq;
  2. import java.util.Collection;
  3. import java.util.Iterator;
  4. import javax.jms.Connection;
  5. import javax.jms.ConnectionFactory;
  6. import javax.jms.Destination;
  7. import javax.jms.MapMessage;
  8. import javax.jms.MessageConsumer;
  9. import javax.jms.Session;
  10. import javax.jms.TextMessage;
  11. import org.apache.activemq.ActiveMQConnection;
  12. import org.apache.activemq.ActiveMQConnectionFactory;
  13. public class Receiver {
  14. public static void main(String[] args) {
  15. // ConnectionFactory :连接工厂,JMS 用它创建连接
  16. ConnectionFactory connectionFactory;
  17. // Connection :JMS 客户端到JMS Provider 的连接
  18. Connection connection = null;
  19. // Session: 一个发送或接收消息的线程
  20. Session session;
  21. // Destination :消息的目的地;消息发送给谁.
  22. Destination destination;
  23. // 消费者,消息接收者
  24. MessageConsumer consumer;
  25. connectionFactory = new ActiveMQConnectionFactory(
  26. ActiveMQConnection.DEFAULT_USER,
  27. ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
  28. try {
  29. // 构造从工厂得到连接对象
  30. connection = connectionFactory.createConnection();
  31. // 启动
  32. connection.start();
  33. // 获取操作连接
  34. session = connection.createSession(Boolean.FALSE,
  35. Session.AUTO_ACKNOWLEDGE);
  36. // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
  37. destination = session.createQueue("FirstQueue");
  38. consumer = session.createConsumer(destination);
  39. // consumer2 = session.createConsumer(destination);
  40. System.out.println("消费者1:消费者程序开始运行...");
  41. while (true) {
  42. // 设置接收者接收消息的时间,为了便于测试,这里谁定为100s
  43. TextMessage message = (TextMessage) consumer.receive(18000);
  44. // Object message = (Object) consumer.receive(28000);
  45. if (null != message) {
  46. System.out.println("消费者1:收到消息-->" + message.getText());
  47. } else {
  48. System.out.println("消费者1:运行结束...\n");
  49. break;
  50. }
  51. }
  52. } catch (Exception e) {
  53. e.printStackTrace();
  54. } finally {
  55. try {
  56. if (null != connection)
  57. connection.close();
  58. } catch (Throwable ignore) {
  59. }
  60. }
  61. }
  62. }

生产者程序:

 
  1. package com.ll.activemq;
  2. import javax.jms.Connection;
  3. import javax.jms.ConnectionFactory;
  4. import javax.jms.DeliveryMode;
  5. import javax.jms.Destination;
  6. import javax.jms.MessageProducer;
  7. import javax.jms.Session;
  8. import javax.jms.TextMessage;
  9. import org.apache.activemq.ActiveMQConnection;
  10. import org.apache.activemq.ActiveMQConnectionFactory;
  11. public class Sender {
  12. private static final int SEND_NUMBER = 5;
  13. public static void main(String[] args) {
  14. // ConnectionFactory :连接工厂,JMS 用它创建连接
  15. ConnectionFactory connectionFactory;
  16. // Connection :JMS 客户端到JMS Provider 的连接
  17. Connection connection = null;
  18. // Session: 一个发送或接收消息的线程
  19. Session session;
  20. // Destination :消息的目的地;消息发送给谁.
  21. Destination destination;
  22. // MessageProducer:消息发送者
  23. MessageProducer producer;
  24. // TextMessage message;
  25. // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar
  26. connectionFactory = new ActiveMQConnectionFactory(
  27. ActiveMQConnection.DEFAULT_USER,
  28. ActiveMQConnection.DEFAULT_PASSWORD,
  29. "tcp://localhost:61616");
  30. try {
  31. // 构造从工厂得到连接对象
  32. connection = connectionFactory.createConnection();
  33. // 启动
  34. connection.start();
  35. // 获取操作连接
  36. session = connection.createSession(Boolean.TRUE,
  37. Session.AUTO_ACKNOWLEDGE);
  38. // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
  39. destination = session.createQueue("FirstQueue");
  40. // 得到消息生成者【发送者】
  41. producer = session.createProducer(destination);
  42. // 设置不持久化,此处学习,实际根据项目决定
  43. producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  44. // 构造消息,此处写死,项目就是参数,或者方法获取
  45. sendMessage(session, producer);
  46. session.commit();
  47. } catch (Exception e) {
  48. e.printStackTrace();
  49. } finally {
  50. try {
  51. if (null != connection)
  52. connection.close();
  53. } catch (Throwable ignore) {
  54. }
  55. }
  56. }
  57. public static void sendMessage(Session session, MessageProducer producer)
  58. throws Exception {
  59. for (int i = 1; i <= SEND_NUMBER; i++) {
  60. TextMessage message = session
  61. .createTextMessage("ActiveMq 发送的消息" + i);
  62. // 发送消息到目的地方
  63. System.out.println("发送消息:" + "ActiveMq 发送的消息" + i);
  64. producer.send(message);
  65. }
  66. System.out.println("生产者程序退出...");
  67. }
  68. }

附件列表

 

转载于:https://www.cnblogs.com/ssslinppp/p/4461495.html

你可能感兴趣的文章
LoadRunner参数化详解
查看>>
P1985 翻转棋
查看>>
python_day1
查看>>
PowerDesigner中CDM数据类型和PDM数据类型间的mapping (对应关系)详解
查看>>
利用HttpWebRequest模拟表单提交
查看>>
面稀土,战码家(一)
查看>>
no server suitable for synchronization found的解决办法
查看>>
MariaDB常用命令
查看>>
SVG.<text/>文字的长度
查看>>
关于二维数组的最大子数组 曹玉松&&蔡迎盈
查看>>
利用反卷积神经网络可视化CNN
查看>>
关于getElementsByTagName的返回值
查看>>
放假第五周
查看>>
App提交审核被拒的原因汇总
查看>>
自加入屠龙后的成长记
查看>>
Java多线程打辅助的三个小伙子
查看>>
【数据挖掘】Clementine(数据挖掘软件)
查看>>
Mysql复制表结构、表数据
查看>>
再看单例模式
查看>>
作为程序员,你关注哪些国外 IT 网站?
查看>>