activemq架构设计 activemq入门

1. ActiveMQ

2.

ActiveMQ工作原理(JSM java message server)

1.  解决服务之间的耦合
2.  使用消息队列,增加系统并发处理量

3.ActiveMQ

时候使用 MQ， 将发短信、发邮箱，通知 MQ，由另外服务平台完成

2、 搜索平台、缓存平台

• 查询数据，建立缓存、索引 ，不从数据库查询，从缓存或者索引库查询
• 当增加、修改、删除数据时，发送消息给 MQ， 缓存平台、索引平台 从 MQ 获取到这个信息，更新缓存或者索引

ActiveMQ的使用

• 第一步:进入 apache-activemq-5.14.0\bin\win64 目录 启动 activemq.bat 文件
• 第二步:访问：http://localhost:8161/ 用户名和密码 都是 admin

ActiveMQ

• 有两种数据结构 Queue、Topic

• 1、 Queue 队列 ，生产者生产了一个消息，只能由一个消费者进行消费
• 2、 Topic 话题，生产者生产了一个消息，可以由多个消费者进行消费

使用Java程序操作ActiveMQ

• 第一步:导入activeMQ的坐标
• 第二步:(一般不用自己写)编写MQ消息生产者

@Test
  
        public    void    testProducerMQ   ()  
  throws    Exception {
  
              // 连接工程-依赖   jms    java中的消息接口(默认的用户名密码,路径)
  
              // 路径写法是    tcp   ://host:port
  
           ConnectionFactory    connectionFactory    =  
  new    ActiveMQConnectionFactory();
  
           // 获取一个链接
  
           Connection    connection    =  
  connectionFactory 
  .createConnection();
  
           // 建立会话
  
           Session    session    =  
  connection 
  .createSession( 
  true   , Session.   AUTO_ACKNOWLEDGE   ); 
  
/其中第一个参数是(是否开启事务,开启后只有在commit的时候才发),
  
            第二个参数代表自动签收,(当commit的时候就签收)
  
           // 创建队列,话题对象
  
           Queue    queue    =  
  session 
  .createQueue( 
  "HelloWorld" 
  );
  
           // 创建消息生产者or 消费者
  
           MessageProducer    producer    =  
  session 
  .createProducer( 
  queue 
  );
  
           // 发送消息
  
              for    (   int    i    = 0;  
  i  
  < 10;  
  i 
  ++) {
  
                producer.send(session.createTextMessage("你好,activeMQ"));
  
           }
  
           // 提交消息
  
              session.commit();
  
     }

默认 tcp 连接 activeMQ

@Test
  
           public void testConsume() throws Exception{
  
                // 连接工程-依赖jms java中的消息接口(默认的用户名密码,路径)
  
                // 路径写法是 tcp://host:port
  
                ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
  
                // 获取一个链接
  
                Connection connection = connectionFactory.createConnection();
  
                //开启链接\
  
                   connection.start();
  
                // 建立会话
  
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  
                // 创建队列,话题对象
  
                Queue queue = session.createQueue("HelloWorld");
  
                // 创建消息生产者or 消费者
  
                MessageConsumer consumer = session.createConsumer(queue);
  
                
  
                //利用监听器接收消息
  
                consumer.setMessageListener(new MessageListener() {
  
                     
  
                     public void onMessage(Message message) {
  
                           TextMessage message2 = (TextMessage) message;
  
                           try {
  
                                System.out.println(message2.getText());
  
                           } catch (JMSException e) {
  
                                // TODO Auto-generated catch block
  
                                e.printStackTrace();
  
                           }
  
                           
  
                     }
  
                });

Spring 结合 ActiveMQ编程

1 在activeMQ_spring导入相关jar包

下图为activemq依赖包

下图为spring 结合activemq的依赖包

2 如果是消息生厂商,则需要配置application-mq.xml

applicationContext-mq.xml3.8 KB

另外配置需要的生产者

@Autowired
  
    @Qualifier("jmsQueueTemplate")
  
    private JmsTemplate jmsTemplate;
 
  
 
调用生产者的方法    
 
//调用mq服务,发送一条消息
  
           jmsTemplate.send("bos_sms",new MessageCreator() {
  
                
  
                @Override
  
                public Message createMessage(Session session) throws JMSException {
  
                     MapMessage mapMessage = session.createMapMessage();
  
                     mapMessage.setString("telephone", order.getCourier().getTelephone());
  
                     mapMessage.setString("msg", msg);
  
                     
  
                     return mapMessage;
  
                }
  
           });

3,消费者,直接搭建一个服务器工程

applicationContext-mq-consume...3.1 KB

创建消费者的类

@Service
  
public class SmsConsumer implements MessageListener{
  
     @Override
  
     public void onMessage(Message message) {
  
           
  
           //调用smsutils来发送短信
  
           // String result = SmsUtils.sendSmsByHTTP(model.getTelephone(), msg);
  
           MapMessage mapMessage = (MapMessage) message;
  
           try {
  
                String telephone = mapMessage.getString("telephone");
  
                String msg = mapMessage.getString("msg");
  
                System.out.println("发送短信内容:"+msg+"接收手机号:"+telephone);
  
           } catch (JMSException e) {
  
                // TODO Auto-generated catch block
  
                e.printStackTrace();
  
           }
  
           String result = "000123";
  
           if (result.startsWith("000")) {
  
                //发送成功
  
                System.out.println("发送成功");
  
           } else {
  
                throw new RuntimeException("短信发送失败,信息码:" + result);
  
           }
  
           
  
           
  
     }
  
}