引入
之前，我们完成了单个消息的发送，以及单个消息发送的多线程池化。
这里，我们继续完成批量发送消息的封装。
因为rabbitMq本身是不支持批量发消息的，所以我们可以直接使用上文所创建的连接池来发送。
最简单的代码是这样的：
# ProducerClient.class
@Override
public void send(List<Message> messages) throws MessageRuntimeException {
messages.forEach(message -> {
message.setMessageType(MessageType.RAPID);
});
rabbitBroker.sendMessages(messages);
}
# RabbitBrokerImpl.class
@Override
public void sendMessages(List<Message> messages) {
messages.forEach(message -> {
sendKernel(message);
});
}
这里使用ThreadLocal。为什么要将批量发送的消息存储到ThreadLocal里呢？这样做的好处相比于直接调用rabbitBroker.sendMessages(List<Message> messages)把批量消息直接放到参数变量里，优点是什么？
这样做是一个请求线程里来做，好处就是在请求链路的任何时候都可以往里面put消息，而不是集中在一起put。
什么是ThreadLocal
ThreadLocal一般称为线程本地变量，它是一种特殊的线程绑定机制，将变量与线程绑定在一起，为每一个线程维护一个独立的变量副本。通过ThreadLocal可以将对象的可见范围限制在同一个线程内。
-
举例说明：
对于教师判分来说，每个教师登陆后会从答题记录表中抽取20道学生的答案进行阅卷，为了避免同一道题被多个教师抽到，需要加锁进行控制，保证当时只有一个教师在抽题，其他教师只能等待，只有当这名教师抽题完成，等待的教师才可以进行抽题。同步机制就是为了同步多个线程对相同资源的并发访问，解决了多个线程之间进行通信的问题。
ThreadLocal就从另一个角度来解决多线程的并发访问，ThreadLocal会为每一个线程维护一个和该线程绑定的变量的副本，从而隔离了多个线程的数据，每一个线程都拥有自己的变量副本，从而也就没有必要对该变量进行同步了。最明显的，ThreadLoacl变量的活动范围为某线程，并且我的理解是该线程“专有的，独自霸占”，对该变量的所有操作均有该线程完成！ThreadLocal提供了线程安全的共享对象，在编写多线程代码时，可以把不安全的整个变量封装进ThreadLocal，或者把该对象的特定于线程的状态封装进ThreadLocal。
ThreadLocal是怎么进行线程绑定的
thread里有个全局变量threadlocalmap,范型为<ThreadLocal,Object>，threadlocal在执行set方法时先获取当前线程，拿到threadlocalmap，以以身为key把值放到map中，从而实现变量与线程的绑定。
源码如下：
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}
public T get() {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
return setInitialValue();
}
可以看到内部有一个ThreadLocalMap，每次都会获取当前线程，在从当前线程中插入或拿取值，这样就保证了只能当前线程能插入和拿取自己的值，其他线程是访问不到的。这样也就避免了传参问题。
所以一般做多数据源切换时，会使用到ThreadLocal。
private static ThreadLocal<Connection> connectionHolder = new ThreadLocal<Connection>() {
public Connection initialValue() {
return DriverManager.getConnection(DB_URL);
}
};
public static Connection getConnection() {
return connectionHolder.get();
}
在多数据源中，会频繁的调用数据库，采用传参来决定使用什么数据源肯定是不可取的，采用ThreadLocal完美解决数据库切换问题。缺点就是以空间换时间的方式(与synchronized相反)，以耗费内存为代价，但是大大减少了线程同步(如synchronized)所带来性能消耗以及减少了线程并发控制的复杂度。
关于ThreadLocal以后会做一个详尽的专题讲解，在此不做赘述。
使用ThreadLocal
首先实现一个ThreadLocal类：
public class MessageHolder {
private List<Message> messages= new ArrayList<>();
@SuppressWarnings({"rawtypes","unchecked"})
public static final ThreadLocal<MessageHolder> holder=new ThreadLocal(){
@Override
protected Object initialValue() {
return new MessageHolder();
}
};
public static void add(Message message){
holder.get().messages.add(message);
}
public static List<Message> clear(){
List<Message> tmp=new ArrayList<>(holder.get().messages);
holder.remove();
return tmp;
}
}
我们可以看到，有add()加入方法，也有取出ThreadLocal中的变量的方法clear()。
那么我们的代码可以变为：
# ProducerClient.class
@Override
public void send(List<Message> messages) throws MessageRuntimeException {
messages.forEach(message -> {
message.setMessageType(MessageType.RAPID);
MessageHolder.add(message);
});
rabbitBroker.sendMessages();
}
# RabbitBrokerImpl.class
@Override
public void sendMessages() {
List<Message> messages=MessageHolder.clear();
messages.forEach(message -> {
sendKernel(message);
});
}
很好理解，变量直接存入ThreadLocal。sendMessages()的实现方法直接从ThreadLocal中取值即可。
