飞天班第42节：ActiveMQ消息队列实战-2

import org.apache.activemq.broker.BrokerService;

public class EmbedBroker {
    public static void main(String[] args) throws Exception {
        // 1.创建服务
        BrokerService broker = new BrokerService();

        // 2、配置连接地址
        // jmx是java的一个新框架
        // 允许将我们所有的资源（软件和硬件）封装一个Java对象，直接暴露在服务中即可使用
        broker.setUseJmx(true);
        broker.addConnector("tcp://localhost:61616");

        // 3、启动
        broker.start();
    }
}

import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;

// 注意导包，util java中的队列
// JMS 下的队列
import javax.jms.Queue;

@Configuration
@EnableJms // 声明对 JMS 注解支持
public class ActiveMQConfig {
    @Value("${myqueue}")
    private String myqueue;

    @Bean
    public Queue myqueue(){
        return new ActiveMQQueue(myqueue);
    }
}

@Service
public class QueueProduceService {
	@Autowired
	JmsMessagingTemplate jmsMessagingTemplate;

	@Autowired
	Queue myqueue;

	public void produceMsg() {
		jmsMessagingTemplate.convertAndSend(myqueue,"生产者发送一个消息");
	}
}

@RestController
public class ActiveMQController {
	@Autowired
	QueueProduceService queueProduceService;

	@RequestMapping("/queue/send")
	public String queueSend() {
		queueProduceService.produceMsg();
		return "消息发送成功";
	}
}

	// 每隔3秒 发送一次
	@Scheduled(fixedDelay = 3000)
	public void produceMsgScheduled() {
		jmsMessagingTemplate.convertAndSend(myqueue,"系统定时发送一个消息");
		System.out.println("系统正在定时发送消息.....");
	}

@EnableScheduling
@SpringBootApplication
public class SpringbootActivemqApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpringbootActivemqApplication.class, args);
    }
}

@Service
public class QueueConsumerService {

	// 通过监听注解实现消息接收。
	// 具体消费  目的地：（队列还是主题）  接收到的消息
	@JmsListener(destination= "${myqueue}")
	public void receive(TextMessage textMessage) throws JMSException {
		// 接收消息
		System.out.println(textMessage.getText());
	}
}

// 注意导包，JMS下的主题
import javax.jms.Topic;

@Value("${mytopic}")
private String mytopic;

@Bean
public Topic mytopic(){
  return new ActiveMQTopic(mytopic);
}

@Service
public class TopicProduceService {
	@Autowired
	JmsMessagingTemplate jmsMessagingTemplate;

	@Autowired
	Topic mytopic; // 注意导包，JMS下的主题

	public void produceMsg() {
		jmsMessagingTemplate.convertAndSend(mytopic,"生产者发送一个主题消息");
	}
}

@RequestMapping("/topic/send")
public String topicSend() {
  topicProduceService.produceMsg();
  return "主题消息发送成功";
}

// 启动两个订阅者监听
@Service
public class TopicConsumerService {
	@JmsListener(destination= "${mytopic}")
	public void receive1(TextMessage textMessage) throws JMSException {
		// 接收消息
		System.out.println("订阅者1号：" + textMessage.getText());
	}

	@JmsListener(destination= "${mytopic}")
	public void receive2(TextMessage textMessage) throws JMSException {
		// 接收消息
		System.out.println("订阅者2号：" + textMessage.getText());
	}
}

package com.coding.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class JmsProduce {
    // 1、消息发送到哪里
 	// http://139.199.13.139:8161 控制台
	// tcp://139.199.13.139:61616 程序通信协议
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);

        // 3、获得连接
        Connection connection = factory.createConnection();
        connection.start();

        // 4、创建会话
        // 参数：事务、签收  AUTO_ACKNOWLEDGE ： 自动确认
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5、模式，目的地（队列，主题）
        // session.createTopic();
        Queue queue = session.createQueue("jdbc-queue01");

        // 6、发送消息（生成者）
        MessageProducer producer = session.createProducer(queue);

        // 开启持久化配置,否则不进入数据库
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);

        // 7、发送具体的消息到队列, 创建的消息类型是什么就些什么
        // 简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。
        for (int i = 0; i < 3; i++) {
            TextMessage message = session.createTextMessage("msg=>" + i);
            // 8、发送消息
            producer.send(message); // 积压，但是没有放入队列
        }

        // 9、用完记得关闭资源
        producer.close();
        session.close();
        connection.close();

        System.out.println("发送完毕");
    }
}

package com.coding.queue;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class JmsConsumer {


    // 1、消息发送到哪里
    // 39.105.61.80:8161  控制板
    // 39.105.61.80:61616 程序通信tcp协议
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

    public static void main(String[] args) throws JMSException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        Connection connection = factory.createConnection();
        connection.start();

      // true开启事务，需要session.commit()
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

        // 5、模式，目的地（队列，主题）
        // session.createTopic();
        Queue queue = session.createQueue("jdbc-queue01");

        // 6、消息接受者（角色变化）
        MessageConsumer consumer = session.createConsumer(queue);

        // 7、获取消息
        while (true){
            // tcp  等待接收消息的过程。 receive
            // receive 阻塞等待
            // receive(long timeout) 超过多少ms就消费结束，然后不等待了
            TextMessage receive = (TextMessage) consumer.receive();

            if (receive!=null){
                System.out.println("接收到消息："+ receive.getText());
            }else {
                break;
            }
        }

        session.commit(); // 假设客户端开启了事务，业务提交事务，否则不成功的！

        // 9、用完记得关闭资源
        consumer.close();
        session.close();
        connection.close();

        System.out.println("接收完毕");
    }
}

package com.coding.topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class JmsProduceTopic_Persist {
  // 1、消息发送到哪里
  // 39.105.61.80:8161  控制板
  // 39.105.61.80:61616 程序通信tcp协议
  public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

  public static void main(String[] args) throws JMSException {
    // 1、创建一个连接工厂
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    // 2、创建连接
    Connection connection = factory.createConnection();
   
    // 3、创建会话
    // connection.createSession(是否开启事务，签收机制)
    // true: 先执行send方法，然后在执行 commit 才是真正的放入队列。可以积压多个消息，然后一次发送
    // false:  send。直接进入队列，加入你关闭了事务，签收一定设置。
    Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
   	// 4、创建目的地
    Topic topic = session.createTopic("jdbc-topic01-persist");
    // 5、创建消息生产者
    MessageProducer producer = session.createProducer(topic);
    // 持久化策略开启后再连接(注意即可，消息发送就持久化)
    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    // 6、开启连接
    connection.start();

    // 7、发送具体的消息到队列, 创建的消息类型是什么就些什么
    // 简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。
    for (int i = 0; i < 6; i++) {
      TextMessage message = session.createTextMessage("msg=>" + i);
      producer.send(message);
    }

    session.commit();
    
    // 8、用完记得关闭资源
    producer.close();
    session.close();
    connection.close();

    System.out.println("发送完毕");
  }
}

package com.coding.topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
import java.io.IOException;

public class JmsConsumerTopic_Persist {
  // 1、消息发送到哪里
  // 39.105.61.80:8161  控制板
  // 39.105.61.80:61616 程序通信tcp协议
  public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616";

  public static void main(String[] args) throws JMSException, IOException {
    // 1、创建一个连接工厂
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    // 2、创建连接
    Connection connection = factory.createConnection();
    // 设置连接ID，我是谁（id），要订阅谁（topic-coding-persist），（客户端信息）
    connection.setClientID("icoding");
    // 3、创建会话
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    // 4、创建目的地
    Topic topic = session.createTopic("jdbc-topic01-persist");
    // 5、创建订阅人
    TopicSubscriber subscriber = session.createDurableSubscriber(topic, "学员-jude");
    //TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "coding老师2");
    
    // 6、开启连接
    connection.start(); 

    // 7、获取消息，可以写基本的聊天程序了,接收,会一直阻塞接收
    Message message = topicSubscriber.receive();
    while (message!=null){
      TextMessage textMessage = (TextMessage) message;
      System.out.println("收到持久化的消息："+textMessage.getText());
      // 自动取消订阅，默认永久的
      message = topicSubscriber.receive(5000L);
    }

    // 8、用完记得关闭资源
    session.close();
    connection.close();

    System.out.println("接收完毕");
  }
}

package com.coding.queue;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageProducer;
import org.apache.activemq.AsyncCallback;

import javax.jms.*;
import java.util.UUID;

public class JmsProduce {
  	// 开启异步投递(默认开启)
    public static final String ACTIVEMQ_URL = "tcp://127.0.0.1:61616?jms.useAsyncSend=true";

    public static void main(String[] args) throws JMSException {
        // 2、创建一个连接工厂
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);

        // 3、获得连接
        Connection connection = factory.createConnection();
        connection.start();

        // 4、创建会话
        // 参数：事务、签收  AUTO_ACKNOWLEDGE ： 自动确认
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 5、模式，目的地（队列，主题）
        // session.createTopic();
        Queue queue = session.createQueue("queue01");

        // ((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);
        // 开启异步投递之后，我们可以使用 ActiveMQMessageProducer 来进行回调的接收
        // 6、发送消息（生成者）
        ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);

        // 开启持久化配置
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);

        // 7、发送具体的消息到队列, 创建的消息类型是什么就些什么
        // 简单文本(TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。
        for (int i = 0; i < 3; i++) {
            TextMessage message = session.createTextMessage("msg=>" + i);
            // 自己给消息定义一个消息的id
            message.setJMSMessageID(UUID.randomUUID().toString()+"自定义id");
            String messageID = message.getJMSMessageID();
            // 8、发送消息，回调函数（很重要）。
            producer.send(message, new AsyncCallback() {
                @Override
                public void onSuccess() { // 发送成功！那个消息发送成功了，发送的时候就知道了
                    System.out.println(messageID);
                }

                @Override
                public void onException(JMSException exception) { // 发送失败！在针对这个消息重发即可。
                    // 处理失败的异常即可
                    System.out.println(messageID);
                }
            });
            //producer.send(message); // 这里我们应该要确认消息是发送完毕的，这样才能保证效率。
        }
        // 9、用完记得关闭资源
        producer.close();
        session.close();
        connection.close();

        System.out.println("发送完毕");
    }
}

// 每隔3秒 发送一次
@Scheduled(fixedDelay = 3000)
public void produceMsgScheduled(){
    template.convertAndSend(topic,"produceMsgTopic");
    System.out.println("系统正在定时发送消息.....");
}

TextMessage message = session.createTextMessage("msg=>" + i);
// 消息的延时和定时发送
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *"); // 每分钟发送一次
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 1000);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 1000);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 9);

// 假设进入了死信队列，我们从死信队列取值即可
Queue queue = session.createQueue("ActiveMQ.DLQ");