Learn more about and use ThreadLocal
What is ThreadLocal
This class provides thread-local variables. These variables differ from their normal counterparts in that each thread that accesses one (via its get or set method) has its own, independently initialized copy of the variable. ThreadLocal instances are typically private static fields in classes that wish to associate state with a thread (e.g., a user ID or Transaction ID). For example, the class below generates unique identifiers local to each thread. A thread's id is assigned the first time it invokes ThreadId.get() and remains unchanged on subsequent calls.
ThreadLocal provides the local variables of threads. Each thread can operate this variable through set() and get() methods, and will not conflict with the local variables of other threads, ensuring the isolation of local variables between threads.
ThrealLocal variables belong to the current thread. Even if they are the same ThreadLocal variable in the same method, their values are different under different business operations. Thread safety.
ThreadLocal business application scenario
Database connection
In the era of jdbc, the database connection needs to be maintained manually. We can use the database connection pool to realize the same database connection. However, when connecting to different databases, different threads need to obtain different connections. At this time, ThreadLocal can be used to maintain the connections of different data sources in the thread pool.
Example code:
public class DynamicDataSourceContextHolder {
private static final Logger logger = LoggerFactory.getLogger(DynamicDataSourceContextHolder.class);
/**
* It is used to ensure that it will not be modified by other threads when switching data sources
*/
private static Lock lock = new ReentrantLock();
/**
* Counter for round robin
*/
private static int counter = 0;
/**
* Maintain variable for every thread, to avoid effect other thread
*/
private static final ThreadLocal<Object> CONTEXT_HOLDER = ThreadLocal.withInitial(DataSourceKey.master);
/**
* All DataSource List
*/
public static List<Object> dataSourceKeys = new ArrayList<>();
/**
* The constant slaveDataSourceKeys.
*/
public static List<Object> slaveDataSourceKeys = new ArrayList<>();
/**
* To switch DataSource
*
* @param key the key
*/
public static void setDataSourceKey(String key) {
CONTEXT_HOLDER.set(key);
}
/**
* Use master data source.
*/
public static void useMasterDataSource() {
CONTEXT_HOLDER.set(DataSourceKey.master);
}
/**
* When using a read-only data source, select the data source to use by round robin
*/
public static void useSlaveDataSource() {
lock.lock();
try {
int datasourceKeyIndex = counter % slaveDataSourceKeys.size();
CONTEXT_HOLDER.set(String.valueOf(slaveDataSourceKeys.get(datasourceKeyIndex)));
counter++;
} catch (Exception e) {
logger.error("Switch slave datasource failed, error message is {}", e.getMessage());
useMasterDataSource();
e.printStackTrace();
} finally {
lock.unlock();
}
}
/**
* Get current DataSource
*
* @return data source key
*/
public static String getDataSourceKey() {
return CONTEXT_HOLDER.get();
}
/**
* To set DataSource as default
*/
public static void clearDataSourceKey() {
CONTEXT_HOLDER.remove();
}
/**
* Check if give DataSource is in current DataSource list
*
* @param key the key
* @return boolean boolean
*/
public static boolean containDataSourceKey(String key) {
return dataSourceKeys.contains(key);
}
}The above code is used for data source context configuration and switching data sources.
Global variable transfer
The most typical example is the transmission of user data. When requesting, the user's token is transmitted through the Header, and the token is placed in ThreadLocal through interceptors (filters, AOP) and other methods, so that the relevant methods of the current thread can share the variable and reduce parameter transmission.
Sketch Map:
Example code:
public class ContextHolder {
private static final ThreadLocal<AccountLoginInfoBo> ACCOUNT_LOGIN_INFO_HOLDER = new ThreadLocal<>();
private static final ThreadLocal<String> REQUEST_ID_HOLDER = new ThreadLocal<>();
public static AccountLoginInfoBo getAccountLoginInfo() {
return ACCOUNT_LOGIN_INFO_HOLDER.get();
}
// ... Omit other codes
}Link tracking
Under the microservice architecture, if there is an error in the call between multiple services, using link tracking is a common means. Embed the traceId in the thread variable of the request. According to this traceId, you can find the log of the corresponding request in each business application.
Example code:
public class GeneralInterceptorHandler extends HandlerInterceptorAdapter {
@Override
public boolean preHandle(HttpServletRequest request,
javax.servlet.http.HttpServletResponse response, Object handler) throws Exception {
request.setAttribute("_timestamp", System.currentTimeMillis());
String requestId = RandomStringUtils.randomNumeric(10);
MDC.put("requestId", requestId);
BaseContextHolder.setRequestId(requestId);
String requestUri = request.getRequestURI();
BaseContextHolder.setRequestUrl(requestUri);
log.info("request uri: {}", requestUri);
return super.preHandle(request, response, handler);
}
@Override
public void postHandle(HttpServletRequest request,
javax.servlet.http.HttpServletResponse response, Object handler,
ModelAndView modelAndView) throws Exception {
long begin = (long) request.getAttribute("_timestamp");
long end = System.currentTimeMillis();
log.info("process success. cost {}ms. ", end - begin);
MDC.remove("requestId");
BaseContextHolder.remove();
super.postHandle(request, response, handler, modelAndView);
}
}Example results:
In the above code, in addition to putting the generated request ID into the MDC, it is also put into the ThreadLocal defined in the Context class for direct use in the code. In this way, each request ID can be printed through the log. To find the log in the micro service request link, you only need to use cat XXX Log | Rep 8475913673 can find all link logs of this request.
Analysis of ThreadLocal principle
By the way, learn about the principle of ThreadLocal, and take a look at the implementation of set(), get(), and remove().
set() method
/**
* Sets the current thread's copy of this thread-local variable
* to the specified value. Most subclasses will have no need to
* override this method, relying solely on the {@link #initialValue}
* method to set the values of thread-locals.
*
* @param value the value to be stored in the current thread's copy of
* this thread-local.
*/
public void set(T value) {
// Get current thread
Thread t = Thread.currentThread();
// Get and maintain the ThreadLocalMap data of the current thread variable, a data structure similar to HashMap
ThreadLocalMap map = getMap(t);
// If a Map already exists in the current thread, directly call Map set
if (map != null)
map.set(this, value);
// If there is no map, add a new map first, and then set
else
createMap(t, value);
}Check the source code and find that the ThreadLocalMap class is used in the set() method.
Expand to view the source code of ThreadLocalMap class
static class ThreadLocalMap {
/**
* The entries in this hash map extend WeakReference, using
* its main ref field as the key (which is always a
* ThreadLocal object). Note that null keys (i.e. entry.get()
* == null) mean that the key is no longer referenced, so the
* entry can be expunged from table. Such entries are referred to
* as "stale entries" in the code that follows.
*/
static class Entry extends WeakReference<ThreadLocal<?>> {
/** The value associated with this ThreadLocal. */
Object value;
Entry(ThreadLocal<?> k, Object v) {
super(k);
value = v;
}
}
/**
* The initial capacity -- MUST be a power of two.
*/
private static final int INITIAL_CAPACITY = 16;
/**
* The table, resized as necessary.
* table.length MUST always be a power of two.
*/
private Entry[] table;
/**
* The number of entries in the table.
*/
private int size = 0;
/**
* The next size value at which to resize.
*/
private int threshold; // Default to 0
/**
* Set the resize threshold to maintain at worst a 2/3 load factor.
*/
private void setThreshold(int len) {
threshold = len * 2 / 3;
}
/**
* Increment i modulo len.
*/
private static int nextIndex(int i, int len) {
return ((i + 1 < len) ? i + 1 : 0);
}
/**
* Decrement i modulo len.
*/
private static int prevIndex(int i, int len) {
return ((i - 1 >= 0) ? i - 1 : len - 1);
}
/**
* Construct a new map initially containing (firstKey, firstValue).
* ThreadLocalMaps are constructed lazily, so we only create
* one when we have at least one entry to put in it.
*/
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
table = new Entry[INITIAL_CAPACITY];
int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
table[i] = new Entry(firstKey, firstValue);
size = 1;
setThreshold(INITIAL_CAPACITY);
}
/**
* Construct a new map including all Inheritable ThreadLocals
* from given parent map. Called only by createInheritedMap.
*
* @param parentMap the map associated with parent thread.
*/
private ThreadLocalMap(ThreadLocalMap parentMap) {
Entry[] parentTable = parentMap.table;
int len = parentTable.length;
setThreshold(len);
table = new Entry[len];
for (int j = 0; j < len; j++) {
Entry e = parentTable[j];
if (e != null) {
@SuppressWarnings("unchecked")
ThreadLocal<Object> key = (ThreadLocal<Object>) e.get();
if (key != null) {
Object value = key.childValue(e.value);
Entry c = new Entry(key, value);
int h = key.threadLocalHashCode & (len - 1);
while (table[h] != null)
h = nextIndex(h, len);
table[h] = c;
size++;
}
}
}
}
/**
* Get the entry associated with key. This method
* itself handles only the fast path: a direct hit of existing
* key. It otherwise relays to getEntryAfterMiss. This is
* designed to maximize performance for direct hits, in part
* by making this method readily inlinable.
*
* @param key the thread local object
* @return the entry associated with key, or null if no such
*/
private Entry getEntry(ThreadLocal<?> key) {
int i = key.threadLocalHashCode & (table.length - 1);
Entry e = table[i];
if (e != null && e.get() == key)
return e;
else
return getEntryAfterMiss(key, i, e);
}
/**
* Version of getEntry method for use when key is not found in
* its direct hash slot.
*
* @param key the thread local object
* @param i the table index for key's hash code
* @param e the entry at table[i]
* @return the entry associated with key, or null if no such
*/
private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
Entry[] tab = table;
int len = tab.length;
while (e != null) {
ThreadLocal<?> k = e.get();
if (k == key)
return e;
if (k == null)
expungeStaleEntry(i);
else
i = nextIndex(i, len);
e = tab[i];
}
return null;
}
/**
* Set the value associated with key.
*
* @param key the thread local object
* @param value the value to be set
*/
private void set(ThreadLocal<?> key, Object value) {
// We don't use a fast path as with get() because it is at
// least as common to use set() to create new entries as
// it is to replace existing ones, in which case, a fast
// path would fail more often than not.
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for (Entry e = tab[i];
e != null;
e = tab[i = nextIndex(i, len)]) {
ThreadLocal<?> k = e.get();
if (k == key) {
e.value = value;
return;
}
if (k == null) {
replaceStaleEntry(key, value, i);
return;
}
}
tab[i] = new Entry(key, value);
int sz = ++size;
if (!cleanSomeSlots(i, sz) && sz >= threshold)
rehash();
}
/**
* Remove the entry for key.
*/
private void remove(ThreadLocal<?> key) {
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for (Entry e = tab[i];
e != null;
e = tab[i = nextIndex(i, len)]) {
if (e.get() == key) {
e.clear();
expungeStaleEntry(i);
return;
}
}
}
/**
* Replace a stale entry encountered during a set operation
* with an entry for the specified key. The value passed in
* the value parameter is stored in the entry, whether or not
* an entry already exists for the specified key.
*
* As a side effect, this method expunges all stale entries in the
* "run" containing the stale entry. (A run is a sequence of entries
* between two null slots.)
*
* @param key the key
* @param value the value to be associated with key
* @param staleSlot index of the first stale entry encountered while
* searching for key.
*/
private void replaceStaleEntry(ThreadLocal<?> key, Object value,
int staleSlot) {
Entry[] tab = table;
int len = tab.length;
Entry e;
// Back up to check for prior stale entry in current run.
// We clean out whole runs at a time to avoid continual
// incremental rehashing due to garbage collector freeing
// up refs in bunches (i.e., whenever the collector runs).
int slotToExpunge = staleSlot;
for (int i = prevIndex(staleSlot, len);
(e = tab[i]) != null;
i = prevIndex(i, len))
if (e.get() == null)
slotToExpunge = i;
// Find either the key or trailing null slot of run, whichever
// occurs first
for (int i = nextIndex(staleSlot, len);
(e = tab[i]) != null;
i = nextIndex(i, len)) {
ThreadLocal<?> k = e.get();
// If we find key, then we need to swap it
// with the stale entry to maintain hash table order.
// The newly stale slot, or any other stale slot
// encountered above it, can then be sent to expungeStaleEntry
// to remove or rehash all of the other entries in run.
if (k == key) {
e.value = value;
tab[i] = tab[staleSlot];
tab[staleSlot] = e;
// Start expunge at preceding stale entry if it exists
if (slotToExpunge == staleSlot)
slotToExpunge = i;
cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
return;
}
// If we didn't find stale entry on backward scan, the
// first stale entry seen while scanning for key is the
// first still present in the run.
if (k == null && slotToExpunge == staleSlot)
slotToExpunge = i;
}
// If key not found, put new entry in stale slot
tab[staleSlot].value = null;
tab[staleSlot] = new Entry(key, value);
// If there are any other stale entries in run, expunge them
if (slotToExpunge != staleSlot)
cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}
/**
* Expunge a stale entry by rehashing any possibly colliding entries
* lying between staleSlot and the next null slot. This also expunges
* any other stale entries encountered before the trailing null. See
* Knuth, Section 6.4
*
* @param staleSlot index of slot known to have null key
* @return the index of the next null slot after staleSlot
* (all between staleSlot and this slot will have been checked
* for expunging).
*/
private int expungeStaleEntry(int staleSlot) {
Entry[] tab = table;
int len = tab.length;
// expunge entry at staleSlot
tab[staleSlot].value = null;
tab[staleSlot] = null;
size--;
// Rehash until we encounter null
Entry e;
int i;
for (i = nextIndex(staleSlot, len);
(e = tab[i]) != null;
i = nextIndex(i, len)) {
ThreadLocal<?> k = e.get();
if (k == null) {
e.value = null;
tab[i] = null;
size--;
} else {
int h = k.threadLocalHashCode & (len - 1);
if (h != i) {
tab[i] = null;
// Unlike Knuth 6.4 Algorithm R, we must scan until
// null because multiple entries could have been stale.
while (tab[h] != null)
h = nextIndex(h, len);
tab[h] = e;
}
}
}
return i;
}
/**
* Heuristically scan some cells looking for stale entries.
* This is invoked when either a new element is added, or
* another stale one has been expunged. It performs a
* logarithmic number of scans, as a balance between no
* scanning (fast but retains garbage) and a number of scans
* proportional to number of elements, that would find all
* garbage but would cause some insertions to take O(n) time.
*
* @param i a position known NOT to hold a stale entry. The
* scan starts at the element after i.
*
* @param n scan control: {@code log2(n)} cells are scanned,
* unless a stale entry is found, in which case
* {@code log2(table.length)-1} additional cells are scanned.
* When called from insertions, this parameter is the number
* of elements, but when from replaceStaleEntry, it is the
* table length. (Note: all this could be changed to be either
* more or less aggressive by weighting n instead of just
* using straight log n. But this version is simple, fast, and
* seems to work well.)
*
* @return true if any stale entries have been removed.
*/
private boolean cleanSomeSlots(int i, int n) {
boolean removed = false;
Entry[] tab = table;
int len = tab.length;
do {
i = nextIndex(i, len);
Entry e = tab[i];
if (e != null && e.get() == null) {
n = len;
removed = true;
i = expungeStaleEntry(i);
}
} while ( (n >>>= 1) != 0);
return removed;
}
/**
* Re-pack and/or re-size the table. First scan the entire
* table removing stale entries. If this doesn't sufficiently
* shrink the size of the table, double the table size.
*/
private void rehash() {
expungeStaleEntries();
// Use lower threshold for doubling to avoid hysteresis
if (size >= threshold - threshold / 4)
resize();
}
/**
* Double the capacity of the table.
*/
private void resize() {
Entry[] oldTab = table;
int oldLen = oldTab.length;
int newLen = oldLen * 2;
Entry[] newTab = new Entry[newLen];
int count = 0;
for (int j = 0; j < oldLen; ++j) {
Entry e = oldTab[j];
if (e != null) {
ThreadLocal<?> k = e.get();
if (k == null) {
e.value = null; // Help the GC
} else {
int h = k.threadLocalHashCode & (newLen - 1);
while (newTab[h] != null)
h = nextIndex(h, newLen);
newTab[h] = e;
count++;
}
}
}
setThreshold(newLen);
size = count;
table = newTab;
}
/**
* Expunge all stale entries in the table.
*/
private void expungeStaleEntries() {
Entry[] tab = table;
int len = tab.length;
for (int j = 0; j < len; j++) {
Entry e = tab[j];
if (e != null && e.get() == null)
expungeStaleEntry(j);
}
}
}An entry structure is maintained to maintain the data of the node. Carefully, students should have found that the entry structure inherits the WeakReference. From the construction method, we can see that the Key of ThreadLocalMap is maintained by soft reference.
ThreadLocal.ThreadLocalMap threadLocals = null;
The above line of code is a member variable: * for each thread, a ThreadLocalMap is maintained independently, and a thread can also have multiple ThreadLocal variables*
get() method
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();
}private T setInitialValue() {
T value = initialValue();
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
return value;
}protected T initialValue() {
return null;
}The get() method is relatively simple on the whole. It is pasted with key logic codes. When calling get(), if there is a value, the value will be returned. If there is no value, call setInitialValue() to obtain the value, and the initialized value is null. That is to say, if the ThreadLocal variable is not assigned or remove d after assignment, directly calling get() method will not report an error and will return a null value.
remove() method
public void remove() {
ThreadLocalMap m = getMap(Thread.currentThread());
if (m != null)
m.remove(this);
}private void remove(ThreadLocal<?> key) {
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for (Entry e = tab[i];
e != null;
e = tab[i = nextIndex(i, len)]) {
if (e.get() == key) {
e.clear();
expungeStaleEntry(i);
return;
}
}
}When the remove method is called, it will judge whether ThreadLocalMap exists in the current thread. If it exists, it will call ThreadLocalMap remove(key); Traverse the linked list structure and remove the entry node.
Summary
- Thread maintains a reference to ThreadLocalMap, and the key of ThreadLocalMap is maintained by WeakReference.
- The ThreadLocal variable is assigned by the thread, and the ThreadLocal value is not obtained by the ThreadLocal operation.
ThreadLocal memory leak
Stack maintenance diagram of ThreadLocal in JVM:
(see watermark for image source)
The reference of entry to value is a strong application, and the reference of key is a weak reference.
If an object is only weakly referenced, the object will be reclaimed whenever GC occurs, regardless of whether there is enough memory space.
Then the problem comes. If the method of operating ThreadLocal variable has a high QPS and is madly requested, the set() get() method is called at this time, and the remove method is not called, then when GC occurs. The relationship between entry and ThreadLocal is broken, the Key is recycled, and the value is also strongly connected. In this way, according to the accessibility analysis of garbage collection, value is still reachable, but from a business point of view, this value will never be accessed, resulting in memory leakage.
Therefore, when using ThreadLocal, it is necessary to call the remove method, which must be displayed. Otherwise, there is a problem and troubleshooting is very troublesome.
- The remove method can be called at the end of the interceptor, AOP and filter
Thread context missing in thread pool
ThreadLocal cannot be passed in parent and child threads, so the most common method is to copy the ThreadLocal value in the parent thread to the child thread. Therefore, you will often see the following code:
for(value in valueList){
Future<?> taskResult = threadPool.submit(new BizTask(ContextHolder.get()));//Submit the task and set the copy Context to the child thread
results.add(taskResult);
}
for(result in results){
result.get();//Blocking waiting for task execution to complete
}The submitted task definition is as follows:
class BizTask<T> implements Callable<T> {
private String session = null;
public BizTask(String session) {
this.session = session;
}
@Override
public T call(){
try {
ContextHolder.set(this.session);
// Execute business logic
} catch(Exception e){
//log error
} finally {
ContextHolder.remove(); // Clean up the context of ThreadLocal to avoid context concatenation when threads are reused
}
return null;
}
}Corresponding thread context management class:
class ContextHolder {
private static ThreadLocal<String> localThreadCache = new ThreadLocal<>();
public static void set(String cacheValue) {
localThreadCache.set(cacheValue);
}
public static String get() {
return localThreadCache.get();
}
public static void remove() {
localThreadCache.remove();
}
}Thread pool settings:
ThreadPoolExecutor executorPool
= new ThreadPoolExecutor(20, 40, 30, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(40),
new XXXThreadFactory(), ThreadPoolExecutor.CallerRunsPolicy);
The last parameter controls how to handle the submitted task when the thread pool is full. There are four built-in strategies:
ThreadPoolExecutor.AbortPolicy //Throw an exception directly ThreadPoolExecutor.DiscardPolicy //Discard current task ThreadPoolExecutor.DiscardOldestPolicy //Discard the task at the head of the work queue ThreadPoolExecutor.CallerRunsPolicy //To serial execution
It can be seen that when initializing the thread pool, we specify that if the thread pool is full, the newly submitted task will be transferred to serial execution. There will be a problem with our previous writing method. Contextholder will be called during serial execution remove(); The context of the main thread will also be cleared. Even if the subsequent thread pool continues to work in parallel, the context passed to the child thread is already null, and such problems are difficult to find in the pre test.
Thread context lost in parallel stream
If ThreadLocal encounters parallel streams, many interesting things will happen, such as the following code:
class ParallelProcessor<T> {
public void process(List<T> dataList) {
// Check the parameters first, and omit the length limit first
dataList.parallelStream().forEach(entry -> {
doIt();
});
}
private void doIt() {
String session = ContextHolder.get();
// do something
}
}It is easy to find that this code cannot work as expected during offline testing, because the underlying implementation of parallel flow is also a ForkJoin thread pool. Since it is a thread pool, contextholder What get () may get out is a null. Let's follow this idea and change the code again:\
class ParallelProcessor<T> {
private String session;
public ParallelProcessor(String session) {
this.session = session;
}
public void process(List<T> dataList) {
// Check the parameters first, and omit the length limit first
dataList.parallelStream().forEach(entry -> {
try {
ContextHolder.set(session);
// Business processing
doIt();
} catch (Exception e) {
// log it
} finally {
ContextHolder.remove();
}
});
}
private void doIt() {
String session = ContextHolder.get();
// do something
}
}Can the modified code work? If you are lucky, you will find that there are problems with this change. If you are not lucky, this code runs well offline, and this code goes online smoothly. Soon you'll find some other weird bug s in the system. The reason is that the design of parallel flow is special, and the parent thread may also participate in the scheduling of parallel streamline process pool. If the above process method is executed by the parent thread, the context of the parent thread will be cleaned up. As a result, the context copied to the child thread is null, which also causes the problem of losing the context.
reference material
- A thorough understanding of ThreadLocal's past and present life
- Spring Boot MyBatis dynamic data source switching, multiple data sources, read-write separation
- Count ThreadLocal's three big holes. Memory leakage is only pediatrics
Sharing plan
Blog content will be synchronized to Tencent cloud + community , we invite you to join us: https://cloud.tencent.com/
license agreement
This paper adopts Signature - non commercial use - share 4.0 international in the same way License agreement, please indicate the source for reprint.