ActiveMQ高级特性

cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.useAsyncSend=true");  //使用连接URI配置异步发送


((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);//在ConnectionFactory 级别配置异步发送


((ActiveMQConnection)connection).setUseAsyncSend(true); //在连接级别配置异步发送

@Component
public class QueueProducer {

    @Autowired
    private JmsMessagingTemplate template;

    @Autowired
    private Queue queue;

    @Autowired
    private ActiveMQConnection connection;

    /**
     * 消息同步发送: 在消息发送成功到MQ中之前,Producer端一直处于阻塞状态,消息的发送成功与否可从阻塞为阻塞看出
     * 　　　　　　　　1.强制要求同步发送 2.事务持久化发送时
     * 消息异步发送: 消息发出后,MQ中未收到消息之前宕机了,但Producer没有显示发送成功与否,会认为发送成功
     * 　　　　　　　此时,为避免这种情况-->有异步发送的回调机制,发送成功或失败后会给Producer回调结果
     */
    public void produceMessage() {

        //template.getConnectionFactory().createConnection().createSession();
        /*进行复杂应用设置时,可从template中获取connection,进行原始的操作,但此时使用了pool的session缓存spring.jms.cache.enable默认为true,
　　　　　　返回的Connection为CacheConnectionFactory,无法生成ActiveMQMessageProducer,无法进行send方法的AsyncCallback回调
　　　　　　,所以此时需要创建一个ActiveMQMessageConnection的配置类以支持异步的调用并成功回调*/
        try {
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);
            ObjectMessage message = session.createObjectMessage();
            message.setStringProperty("msgId", "001");
            message.setObject(new User(1, "小明"));

            producer.send(message, new AsyncCallback() {
                @Override
                public void onSuccess() {
                    try {
                        System.out.println("消息发送成功:" + message.getStringProperty("msgId"));
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }

                }

                @Override
                public void onException(JMSException e) {
                    try {
                        System.out.println("出现异常:" + message.getStringProperty("msgId"));
                    } catch (JMSException je) {
                        je.printStackTrace();
                    }
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
        //template.convertAndSend(queue,"message:"+new Date().toLocaleString());

    }

public class ActiveMQConnectionConfig {
    @Value("${spring.activemq.broker-url}")
    private String brokerUrl;

    @Bean
    public ActiveMQConnection activeMQConnection() throws JMSException {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(brokerUrl);
        //启用异步发送消息
        activeMQConnectionFactory.setUseAsyncSend(true);
        ActiveMQConnection connection =
                (ActiveMQConnection) activeMQConnectionFactory.createConnection();
        return connection;

<broker xmlns="http://activemq.apache.org/schema/core"  brokerName="localhost" dataDirectory="${ activemq. data}" schedulerSupport="true" />

message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,10*1000); //延时10秒发送
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,10); //重复投递10次
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD,5*1000); //重复投递的间隔时间:5秒
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *"); //使用Corn表达式