ActiveMQ-API(五)

负载均衡（取模做负载均衡）简单利用线程实现吞吐量栗子实现：

package bhz.mq.action;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Producer {

    //单例模式
    //1：连接工厂
    private ConnectionFactory connectionFactory;
    //2:连接对象
    private Connection connection;
    
    //3:Session对象
    private Session session;
    
    //4:生产者
    private MessageProducer messageProducer;
    
    public Producer(){
        try {
            this.connectionFactory = new ActiveMQConnectionFactory("bhz","bhz",
                    "tcp://localhost:61616");
            this.connection = this.connectionFactory.createConnection();
            this.connection.start();
            this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            this.messageProducer = this.session.createProducer(null);
        } catch (Exception e) {
            e.printStackTrace();    
        }
    }
    
    public Session getSession(){
        return this.session;
    }
    
    public void send1(){
        try {
            Destination destination = this.session.createQueue("first");
            for (int i=0;i<100;i++) {
                MapMessage msg = this.session.createMapMessage();
                int id = i;
                msg.setInt("id", id);
                msg.setString("name", "张"+i);
                msg.setStringProperty("age", ""+i);
                String receiver = id%2==0?"A":"B";
                msg.setStringProperty("receiver", receiver);
                this.messageProducer.send(destination, msg, DeliveryMode.NON_PERSISTENT, 2,1000*60*10L);
                System.out.println("message send id :"+id);
            }
            
            
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public void send2(){
        try {
            Destination destination = this.session.createQueue("first");
            TextMessage msg = this.session.createTextMessage("我是一个字符串内容");
            this.messageProducer.send(destination,msg,DeliveryMode.NON_PERSISTENT,9,1000*60*10L);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) {
        Producer p = new Producer();
        p.send1();
    }
    
    
}

package bhz.mq.action;

import javax.jms.MapMessage;

public class MessageTask implements Runnable{

    private MapMessage message;
    
    public MessageTask(MapMessage message){
        this.message = message;
    }
    
    @Override
    public void run() {
        try {
            Thread.sleep(500);
            System.out.println("当前线程："+Thread.currentThread().getName()+"处理任务："+this.message.getString("id"));
        } catch (Exception e) {
            e.printStackTrace();
        }
        
    }

}

package bhz.mq.action;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerA {

    public final String SELECTOR="receiver = 'A'";
    
    //1：连接工厂
    private ConnectionFactory connectionFactory;
    //2:连接对象
    private Connection connection;
        
    //3:Session对象
    private Session session;
    
    //4:消费者
    private MessageConsumer messageConsumer;
    
    //5：目标地址
    private Destination destination;
    
    public ConsumerA(){
        try {
            this.connectionFactory = new ActiveMQConnectionFactory("bhz","bhz",
                    "tcp://localhost:61616");
            this.connection = this.connectionFactory.createConnection();
            this.connection.start();
            this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            this.destination=this.session.createQueue("first");
            //创建消费者的时候发生了变化
            this.messageConsumer = this.session.createConsumer(this.destination, SELECTOR);
            System.out.println("Consumer A start...");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public void receiver(){
        try {
            this.messageConsumer.setMessageListener(new Listener());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    class Listener implements MessageListener{

        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10000);
        ExecutorService executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),20,120L,TimeUnit.SECONDS,queue);
        
        @Override
        public void onMessage(Message message) {
            try {
                if (message instanceof TextMessage) {
                    
                }
                if (message instanceof MapMessage) {
                    MapMessage ret = (MapMessage)message;
                    executor.execute(new MessageTask(ret));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            
        }
        
    }
    
    public static void main(String[] args) {
        ConsumerA c = new ConsumerA();
        c.receiver();
    }
    
}

package bhz.mq.action;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerB {

    public final String SELECTOR="receiver = 'B'";
    
    //1：连接工厂
    private ConnectionFactory connectionFactory;
    //2:连接对象
    private Connection connection;
        
    //3:Session对象
    private Session session;
    
    //4:消费者
    private MessageConsumer messageConsumer;
    
    //5：目标地址
    private Destination destination;
    
    public ConsumerB(){
        try {
            this.connectionFactory = new ActiveMQConnectionFactory("bhz","bhz",
                    "tcp://localhost:61616");
            this.connection = this.connectionFactory.createConnection();
            this.connection.start();
            this.session = this.connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
            this.destination=this.session.createQueue("first");
            //创建消费者的时候发生了变化
            this.messageConsumer = this.session.createConsumer(this.destination, SELECTOR);
            System.out.println("Consumer B start...");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public void receiver(){
        try {
            this.messageConsumer.setMessageListener(new Listener());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    class Listener implements MessageListener{

        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10000);
        ExecutorService executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),20,120L,TimeUnit.SECONDS,queue);
        
        @Override
        public void onMessage(Message message) {
            try {
                if (message instanceof TextMessage) {
                    
                }
                if (message instanceof MapMessage) {
                    MapMessage ret = (MapMessage)message;
                    executor.execute(new MessageTask(ret));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            
        }
        
    }
    
    public static void main(String[] args) {
        ConsumerB c = new ConsumerB();
        c.receiver();
    }
    
}