Springboot@Async开启异步线程

一。创建线程池配置类

package com.example.demo.config;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

import java.util.concurrent.ThreadPoolExecutor;

@Configuration

@EnableAsync

public class TaskPoolConfig {

@Bean("taskExecutor")

public ExecutortaskExecutor(){

ThreadPoolTaskExecutor executor =new ThreadPoolTaskExecutor();

        executor.setCorePoolSize(10);

        executor.setMaxPoolSize(20);

        executor.setQueueCapacity(200);

        executor.setKeepAliveSeconds(60);

        executor.setThreadNamePrefix("taskExecutor-");

        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        executor.setWaitForTasksToCompleteOnShutdown(true);

        executor.setAwaitTerminationSeconds(60);

        return executor;

    }

}

二。用@Async（"taskExecutor"）调用bean

package com.example.demo.roundone;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

import org.springframework.scheduling.annotation.Async;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.stereotype.Component;

import javax.jms.*;

@Component

@EnableAsync

public class Topic {

@Async("taskExecutor")

public  void producer(int times, int interval, String topicname, String url, String context, String username, String password) {

try {

//创建连接工厂

            ConnectionFactory connectionFactory =new ActiveMQConnectionFactory(username, password, url);

            //创建连接

            Connection connection = connectionFactory.createConnection();

            connection.start();

            //创建会话

            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);

            //创建目的地

            Destination destination = session.createTopic(topicname);

            //创建生产者

            MessageProducer messageProducer = session.createProducer(destination);

            for (int i =1; i < times +1; i++) {

//创建消息

                TextMessage textMessage = session.createTextMessage(context + i);

                messageProducer.send(textMessage);

                System.out.println("发送消息：" + textMessage.getText());

                session.commit();

                Thread.sleep(interval);

            }

if (connection !=null) {

connection.close();

            }

}catch (Exception e) {

e.printStackTrace();

        }

}

@Async("taskExecutor")

public  void consumer(String topicname) {

try {

ConnectionFactory connectionFactory =new ActiveMQConnectionFactory("user",

                    "user", "failover:(tcp://localhost:61616)?Randomize=false");

            Connection connection = connectionFactory.createConnection();

            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            Destination destination = session.createTopic(topicname);

            //MessageProducer messageProducer=session.createProducer(null);

            MessageConsumer messageConsumer = session.createConsumer(destination);

            TextMessage textMessage1 = (TextMessage) messageConsumer.receive();

//            messageConsumer.setMessageListener(new MessageListener() {

//                @Override

//                public void onMessage(Message message) {

//                    try {

//                        TextMessage textMessage = (TextMessage) messageConsumer.receive();

//

//                        if (textMessage == null) {

//

//                            System.out.println("error");

//

//                        }

//

//                        System.out.println("收到的内容：" + textMessage.getText());

//                    } catch (JMSException e) {

//                        e.printStackTrace();

//                    }

//                }

//            });

            while(true){

TextMessage textMessage=(TextMessage)messageConsumer.receive();

                if(textMessage==null){

break;

                }

System.out.println("收到的内容："+textMessage.getText());

            }

}catch (Exception e) {

e.printStackTrace();

        }

}

}

三。测试

package com.example.demo;

import com.example.demo.config.TaskPoolConfig;

import com.example.demo.roundone.Topic;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)

@SpringBootTest

public class DemoApplicationTests {

@Autowired

    Topic topic;

    @Test

    public void test() {

topic.producer(5,500,"testone","tcp://localhost:61616","this is a test from lyq","user","user");

        topic.consumer("testone");

        }

@Test

    public void contextLoads() {

}

}