ActiveMQ 知识点

Posted 2022-08-25 loveer

消息队列高可用

持久化，事务，签收，zookeeper+replicated-leveldb-store的主从集群

异步发送

同步发送：

1. 明确指定同步发送
2. 未使用事务的前提下，发送持久化消息（会使用同步发送，且会阻塞producer直到broker返回一个确认，表示消息已持久化。会阻塞客户端带来延时）

异步发送：提高发送效率，但会增加broker性能消耗，且不能有效确保消息发送成功。

开启方式http://activemq.apache.org/async-sends

ActiveMQ默认使用异步发送：非持久化消息、事务内的消息均采用异步发送；对于持久化消息，则采用同步发送。

如果指定了useAsyncSend=true，都将采用异步发送，包括PERSISTENT类型的消息，也将使用异步发送。在useAsyncSend的情况下，客户端需要容忍消息丢失的可能。

正确的异步发送方法：接收回调
接收回执，由客户端再次判断是否发送成功

        ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        TextMessage message = null;
        for (int i = 0; i < count; i++) 
            message = session.createTextMessage("Hello-"+i)
            message.setJMSmessageID(UUID.randomUUID().toString());
            String msgID = message.getJMSmessageID();
            producer.send(message , new AsyncCallback() 
                @Override
                public void onSuccess() 
                    System.out.println(msgID+"成功");
                

                @Override
                public void onException(JMSException exception) 
                    System.out.println(msgID+"失败");
                    exception.printStackTrace();
                
            );
        

1. 同步调用
    是指从请求的发起一直到最终的处理完成期间，请求的调用方一直在同步阻塞等待调用的处理完成。

ClientCode在send发出请求后，就一直都阻塞在这里，调用Service以后，Service会调用Adapter这样一个类来进行处理，
而这个类会调用远程的一个服务等待最终调用结果的返回，是成功还是失败。
因为这个过程是阻塞等待的，所以这个过程也就是同步调用。

2.异步调用
    是指在请求发起的处理过程中，客户端的代码已经返回了，
    它可以继续进行自己的后续操作，而不需要等待调用处理完成，这就叫做异步调用。

异步调用过程，用户Clientcode在send发出请求后，调用Service以后，Service会把这个调用请求发送给消息队列，然后就立即返回了。
Clientcode收到返回以后继续向下处理，不会继续阻塞等待。
在这个过程中，客户端的调用，也就是应用程序的调用，和业务逻辑真正发送邮件的操作是不同步的。

延时 定时投递

官方配置http://activemq.apache.org/delay-and-schedule-message-delivery

默认不开启：Xml配置中将broker的schedulerSupport属性设置为true来启用它

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
long delay = 30 * 1000;    //首次延时时间
long period = 10 * 1000;    //每次间隔时间
int repeat = 9;    //重复
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
producer.send(message);