Java多线程上下文传递在复杂场景下的实践

　　一、引言
　　海外商城从印度做起，慢慢的会有一些其他国家的诉求，这个时候需要我们针对当前的商城做一个改造，可以支撑多个国家的商城，这里会涉及多个问题，多语言，多国家，多时区，本地化等等。在多国家的情况下如何把识别出来的国家信息传递下去，一层一层直到代码执行的最后一步。甚至还有一些多线程的场景需要处理。二、背景技术
　　2。1ThreadLocal
　　ThreadLocal是最容易想到了，入口识别到国家信息后，丢进ThreadLocal，这样后续代码、redis、DB等做国家区分的时候都能使用到。
　　这里先简单介绍一下ThreadLocal：Setsthecurrentthreadscopyofthisthreadlocalvariabletothespecifiedvalue。Mostsubclasseswillhavenoneedtooverridethismethod，relyingsolelyonthe｛linkinitialValue｝methodtosetthevaluesofthreadlocals。paramvaluethevaluetobestoredinthecurrentthreadscopyofthisthreadlocal。publicvoidset（Tvalue）｛ThreadtThread。currentThread（）；ThreadLocalMapmapgetMap（t）；if（map！null）map。set（this，value）；elsecreateMap（t，value）；｝Returnsthevalueinthecurrentthreadscopyofthisthreadlocalvariable。Ifthevariablehasnovalueforthecurrentthread，itisfirstinitializedtothevaluereturnedbyaninvocationofthe｛linkinitialValue｝method。returnthecurrentthreadsvalueofthisthreadlocalpublicTget（）｛ThreadtThread。currentThread（）；ThreadLocalMapmapgetMap（t）；if（map！null）｛ThreadLocalMap。Entryemap。getEntry（this）；if（e！null）｛SuppressWarnings（unchecked）Tresult（T）e。｝｝returnsetInitialValue（）；｝GetthemapassociatedwithaThreadLocal。OverriddeninInheritableThreadLocal。paramtthecurrentthreadreturnthemapThreadLocalMapgetMap（Threadt）｛returnt。threadL｝Gettheentryassociatedwithkey。Thismethoditselfhandlesonlythefastpath：adirecthitofexistingkey。ItotherwiserelaystogetEntryAfterMiss。Thisisdesignedtomaximizeperformancefordirecthits，inpartbymakingthismethodreadilyinlinable。paramkeythethreadlocalobjectreturntheentryassociatedwithkey，ornullifnosuchprivateEntrygetEntry（ThreadL？key）｛intikey。threadLocalHashCode（table。length1）；Entryetable〔i〕；if（e！nulle。get（）key）elsereturngetEntryAfterMiss（key，i，e）；｝每一个Thread线程都有属于自己的threadLocals（ThreadLocalMap），里面有一个弱引用的Entry（ThreadLocal，Object）。get方法首先通过Thread。currentThread得到当前线程，然后拿到线程的threadLocals（ThreadLocalMap），再从Entry中取得当前线程存储的value。set值的时候更改当前线程的threadLocals（ThreadLocalMap）中Entry对应的value值。
　　实际使用中除了同步方法之外，还有起异步线程处理的场景，这个时候就需要把ThreadLocal的内容从父线程传递给子线程，这个怎么办呢？
　　不急，Java还有InheritableThreadLocal来帮我们解决这个问题。
　　2。2InheritableThreadLocapublicclassInheritableThreadLocalTextendsThreadLocalT｛Computesthechildsinitialvalueforthisinheritablethreadlocalvariableasafunctionoftheparentsvalueatthetimethechildthreadiscreated。Thismethodiscalledfromwithintheparentthreadbeforethechildisstarted。pThismethodmerelyreturnsitsinputargument，andshouldbeoverriddenifadifferentbehaviorisdesired。paramparentValuetheparentthreadsvaluereturnthechildthreadsinitialvalueprotectedTchildValue（TparentValue）｛returnparentV｝GetthemapassociatedwithaThreadLocal。paramtthecurrentthreadThreadLocalMapgetMap（Threadt）｛returnt。inheritableThreadL｝CreatethemapassociatedwithaThreadLocal。paramtthecurrentthreadparamfirstValuevaluefortheinitialentryofthetable。voidcreateMap（Threadt，TfirstValue）｛t。inheritableThreadLocalsnewThreadLocalMap（this，firstValue）；｝｝java。lang。Threadinit（java。lang。ThreadGroup，java。lang。Runnable，java。lang。String，long，java。security。AccessControlContext，boolean）if（inheritThreadLocalsparent。inheritableThreadLocals！null）this。inheritableThreadLocalsThreadLocal。createInheritedMap（parent。inheritableThreadLocals）；InheritableThreadLocal操作的是inheritableThreadLocals这个变量，而不是ThreadLocal操作的threadLocals变量。创建新线程的时候会检查父线程中parent。inheritableThreadLocals变量是否为null，如果不为null则复制一份parent。inheritableThreadLocals的数据到子线程的this。inheritableThreadLocals中去。因为复写了getMap（Thread）和CreateMap（）方法直接操作inheritableThreadLocals，这样就实现了在子线程中获取父线程ThreadLocal值。
　　现在在使用多线程的时候，都是通过线程池来做的，这个时候用InheritableThreadLocal可以吗？会有什么问题吗？先看下下面的代码的执行情况：teststaticInheritableThreadLocalStringinheritableThreadLocalnewInheritableThreadLocal（）；publicstaticvoidmain（String〔〕args）throwsInterruptedException｛ExecutorServiceexecutorServiceExecutors。newFixedThreadPool（1）；inheritableThreadLocal。set（iamainheritparent）；executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛System。out。println（inheritableThreadLocal。get（））；｝｝）；TimeUnit。SECONDS。sleep（1）；inheritableThreadLocal。set（iamanewinheritparent）；设置新的值executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛System。out。println（inheritableThreadLocal。get（））；｝｝）；｝iamainheritparentiamainheritparentpublicstaticvoidmain（String〔〕args）throwsInterruptedException｛ExecutorServiceexecutorServiceExecutors。newFixedThreadPool（1）；inheritableThreadLocal。set（iamainheritparent）；executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛System。out。println（inheritableThreadLocal。get（））；inheritableThreadLocal。set（iamaoldinheritparent）；子线程中设置新的值｝｝）；TimeUnit。SECONDS。sleep（1）；inheritableThreadLocal。set（iamanewinheritparent）；主线程设置新的值executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛System。out。println（inheritableThreadLocal。get（））；｝｝）；｝iamainheritparentiamaoldinheritparent
　　这里看第一个执行结果，发现主线程第二次设置的值，没有改掉，还是第一次设置的值iamainheritparent，这是什么原因呢？
　　再看第二个例子的执行结果，发现在第一个任务中设置的iamaoldinheritparent的值，在第二个任务中打印出来了。这又是什么原因呢？
　　回过头来看看上面的源码，在线程池的情况下，第一次创建线程的时候会从父线程中copyinheritableThreadLocals中的数据，所以第一个任务成功拿到了父线程设置的iamainheritparent，第二个任务执行的时候复用了第一个任务的线程，并不会触发复制父线程中的inheritableThreadLocals操作，所以即使在主线程中设置了新的值，也会不生效。同时get（）方法是直接操作inheritableThreadLocals这个变量的，所以就直接拿到了第一个任务设置的值。
　　那遇到线程池应该怎么办呢？
　　2。3TransmittableThreadLocal
　　TransmittableThreadLocal（TTL）这个时候就派上用场了。这是阿里开源的一个组件，我们来看看它怎么解决线程池的问题，先来一段代码，在上面的基础上修改一下，使用TransmittableThreadLocal。staticTransmittableThreadLocalStringtransmittableThreadLocalnewTransmittableThreadLocal（）；使用TransmittableThreadLocalpublicstaticvoidmain（String〔〕args）throwsInterruptedException｛ExecutorServiceexecutorServiceExecutors。newFixedThreadPool（1）；executorServiceTtlExecutors。getTtlExecutorService（executorService）；用TtlExecutors装饰线程池transmittableThreadLocal。set（iamatransmittableparent）；executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛System。out。println（transmittableThreadLocal。get（））；transmittableThreadLocal。set（iamaoldtransmittableparent）；子线程设置新的值｝｝）；System。out。println（transmittableThreadLocal。get（））；TimeUnit。SECONDS。sleep（1）；transmittableThreadLocal。set（iamanewtransmittableparent）；主线程设置新的值executorService。execute（newRunnable（）｛Overridepublicvoidrun（）｛System。out。println（transmittableThreadLocal。get（））；｝｝）；｝iamatransmittableparentiamatransmittableparentiamanewtransmittableparent
　　执行代码后发现，使用TransmittableThreadLocalTtlExecutors。getTtlExecutorService（executorService）装饰线程池之后，在每次调用任务的时，都会将当前的主线程的TransmittableThreadLocal数据copy到子线程里面，执行完成后，再清除掉。同时子线程里面的修改回到主线程时其实并没有生效。这样可以保证每次任务执行的时候都是互不干涉的。这是怎么做到的呢？来看源码。TtlExecutors和TransmittableThreadLocal源码privateTtlRunnable（Runnablerunnable，booleanreleaseTtlValueReferenceAfterRun）｛this。capturedRefnewAtomicReferenceObject（capture（））；this。this。releaseTtlValueReferenceAfterRunreleaseTtlValueReferenceAfterR｝com。alibaba。ttl。TtlRunnablerunwrapmethod｛linkRunnablerun（）｝。Overridepublicvoidrun（）｛ObjectcapturedcapturedRef。get（）；获取线程的ThreadLocalMapif（capturednullreleaseTtlValueReferenceAfterRun！capturedRef。compareAndSet（captured，null））｛thrownewIllegalStateException（TTLvaluereferenceisreleasedafterrun！）；｝Objectbackupreplay（captured）；暂存当前子线程的ThreadLocalMap到backuptry｛runnable。run（）；｝finally｛restore（backup）；恢复线程执行时被改版的Threadlocal对应的值｝｝com。alibaba。ttl。TransmittableThreadLocal。TransmitterreplayReplaythecaptured｛linkTransmittableThreadLocal｝valuesfrom｛linkcapture（）｝，andreturnthebackup｛linkTransmittableThreadLocal｝valuesincurrentthreadbeforereplay。paramcapturedcaptured｛linkTransmittableThreadLocal｝valuesfromotherthreadfrom｛linkcapture（）｝returnthebackup｛linkTransmittableThreadLocal｝valuesbeforereplayseecapture（）since2。3。0publicstaticObjectreplay（Objectcaptured）｛SuppressWarnings（unchecked）MapTransmittableThreadL？，ObjectcapturedMap（MapTransmittableThreadL？，Object）MapTransmittableThreadL？，ObjectbackupnewHashMapTransmittableThreadL？，Object（）；for（I？extendsMap。EntryTransmittableThreadL？，？iteratorholder。get（）。entrySet（）。iterator（）；iterator。hasNext（）；）｛Map。EntryTransmittableThreadL？，？nextiterator。next（）；TransmittableThreadL？threadLocalnext。getKey（）；backupbackup。put（threadLocal，threadLocal。get（））；cleartheTTLvalueonlyincapturedavoidextraTTLvalueincaptured，whenruntask。if（！capturedMap。containsKey（threadLocal））｛iterator。remove（）；threadLocal。superRemove（）；｝｝setvaluetocapturedTTLfor（Map。EntryTransmittableThreadL？，Objectentry：capturedMap。entrySet（））｛SuppressWarnings（unchecked）TransmittableThreadLocalObjectthreadLocal（TransmittableThreadLocalObject）entry。getKey（）；threadLocal。set（entry。getValue（））；｝callbeforeExecutecallbackdoExecuteCallback（true）；｝com。alibaba。ttl。TransmittableThreadLocal。TransmitterrestoreRestorethebackup｛linkTransmittableThreadLocal｝valuesfrom｛linkTransmitterreplay（Object）｝。parambackupthebackup｛linkTransmittableThreadLocal｝valuesfrom｛linkTransmitterreplay（Object）｝since2。3。0publicstaticvoidrestore（Objectbackup）｛SuppressWarnings（unchecked）MapTransmittableThreadL？，ObjectbackupMap（MapTransmittableThreadL？，Object）callafterExecutecallbackdoExecuteCallback（false）；for（I？extendsMap。EntryTransmittableThreadL？，？iteratorholder。get（）。entrySet（）。iterator（）；iterator。hasNext（）；）｛Map。EntryTransmittableThreadL？，？nextiterator。next（）；TransmittableThreadL？threadLocalnext。getKey（）；cleartheTTLvalueonlyinbackupavoidtheextravalueofbackupafterrestoreif（！backupMap。containsKey（threadLocal））｛iterator。remove（）；threadLocal。superRemove（）；｝｝restoreTTLvaluefor（Map。EntryTransmittableThreadL？，Objectentry：backupMap。entrySet（））｛SuppressWarnings（unchecked）TransmittableThreadLocalObjectthreadLocal（TransmittableThreadLocalObject）entry。getKey（）；threadLocal。set（entry。getValue（））；｝｝
　　可以看下整个过程的完整时序图：
　　OK，既然问题都解决了，来看看实际使用吧，有两种使用，先看第一种，涉及HTTP请求、Dubbo请求和job，采用的是数据级别的隔离。三、TTL在海外商城的实际应用
　　3。1不分库，分数据行SpringMVC
　　用户HTTP请求，首先我们要从url或者cookie中解析出国家编号，然后在TransmittableThreadLocal中存放国家信息，在MyBatis的拦截器中读取国家数据，进行sql改造，最终操作指定的国家数据，多线程场景下用TtlExecutors包装原有自定义线程池，保障在使用线程池的时候能够正确将国家信息传递下去。HTTP请求publicclassShopShardingHelperUtil｛privatestaticTransmittableThreadLocalStringcountrySetnewTransmittableThreadLocal（）；获取threadLocal中设置的国家标志returnpublicstaticStringgetCountry（）｛returncountrySet。get（）；｝设置threadLocal中设置的国家publicstaticvoidsetCountry（Stringcountry）｛countrySet。set（country。toLowerCase（））；｝清除标志publicstaticvoidclear（）｛countrySet。remove（）；｝｝拦截器对cookie和url综合判断国家信息，放入到TransmittableThreadLocal中设置线程中的国家标志StringcountrylocaleContext。getLocale（）。getCountry（）。toLowerCase（）；ShopShardingHelperUtil。setCountry（country）；自定义线程池，用TtlExecutors包装原有自定义线程池publicstaticExecutorgetExecutor（）｛if（executornull）｛synchronized（TransmittableExecutor。class）｛if（executornull）｛executorTtlExecutors。getTtlExecutor（initExecutor（））；用TtlExecutors装饰Executor，结合TransmittableThreadLocal解决异步线程threadlocal传递问题｝｝｝｝实际使用线程池的地方，直接调用执行即可TransmittableExecutor。getExecutor（）。execute（newBatchExeRunnable（param1，param2））；mybatis的Interceptor代码，使用TransmittableThreadLocal的国家信息，改造原有sql，加上国家参数，在增删改查sql中区分国家数据publicObjectintercept（Invocationinvocation）throwsThrowable｛StatementHandlerstatementHandler（StatementHandler）invocation。getTarget（）；BoundSqlboundSqlstatementHandler。getBoundSql（）；StringoriginalSqlboundSql。getSql（）；Statementstatement（Statement）CCJSqlParserUtil。parse（originalSql）；StringthreadCountryShopShardingHelperUtil。getCountry（）；线程中的国家不为空才进行处理if（StringUtils。isNotBlank（threadCountry））｛if（statementinstanceofSelect）｛SelectselectStatement（Select）VivoSelectVisitorvivoSelectVisitornewVivoSelectVisitor（threadCountry）；vivoSelectVisitor。init（selectStatement）；｝elseif（statementinstanceofInsert）｛InsertinsertStatement（Insert）VivoInsertVisitorvivoInsertVisitornewVivoInsertVisitor（threadCountry）；vivoInsertVisitor。init（insertStatement）；｝elseif（statementinstanceofUpdate）｛UpdateupdateStatement（Update）VivoUpdateVisitorvivoUpdateVisitornewVivoUpdateVisitor（threadCountry）；vivoUpdateVisitor。init（updateStatement）；｝elseif（statementinstanceofDelete）｛DeletedeleteStatement（Delete）VivoDeleteVisitorvivoDeleteVisitornewVivoDeleteVisitor（threadCountry）；vivoDeleteVisitor。init（deleteStatement）；｝FieldboundSqlFieldBoundSql。class。getDeclaredField（sql）；boundSqlField。setAccessible（true）；boundSqlField。set（boundSql，statement。toString（））；｝else｛logger。error（interceptnotaddcountrysql。。。。statement。toString（））；｝logger。info（interceptquerynewsql。。。。statement。toString（））；调用方法，实际上就是拦截的方法Objectresultinvocation。proceed（）；｝
　　对于Dubbo接口和无法判断国家信息的HTTP接口，在入参部分增加国家信息参数，通过拦截器或者手动set国家信息到TransmittableThreadLocal。
　　对于定时任务job，因为所有国家都需要执行，所以会把所有国家进行遍历执行，这也可以通过简单的注解来解决。
　　这个版本的改造，点检测试也基本通过了，自动化脚本验证也是没问题的，不过因为业务发展问题最终没上线。
　　3。2分库SpringBoot
　　后续在建设新的国家商城的时候，分库分表方案调整为每个国家独立数据库，同时整体开发框架升级到SpringBoot，我们把这套方案做了升级，总体思路是一样的，只是在实现细节上略有不同。
　　SpringBoot里面的异步一般通过Async这个注解来实现，通过自定义线程池来包装，使用时在HTTP请求判断locale信息的写入国家信息，后续完成切DB的操作。
　　对于Dubbo接口和无法判断国家信息的HTTP接口，在入参部分增加国家信息参数，通过拦截器或者手动set国家信息到TransmittableThreadLocal。BeanpublicThreadPoolTaskExecutorthreadPoolTaskExecutor（）｛returnTtlThreadPoolExecutors。getAsyncExecutor（）；｝publicclassTtlThreadPoolExecutors｛privatestaticfinalStringCOMMONBUSINESSCOMMONEXECUTOR；publicstaticfinalintQUEUECAPACITY20000；publicstaticExecutorServicegetExecutorService（）｛returnTtlExecutorServiceMananger。getExecutorService（COMMONBUSINESS）；｝publicstaticExecutorServicegetExecutorService（StringthreadGroupName）｛returnTtlExecutorServiceMananger。getExecutorService（threadGroupName）；｝publicstaticThreadPoolTaskExecutorgetAsyncExecutor（）｛用TtlExecutors装饰Executor，结合TransmittableThreadLocal解决异步线程threadlocal传递问题returngetTtlThreadPoolTaskExecutor（initTaskExecutor（））；｝privatestaticThreadPoolTaskExecutorinitTaskExecutor（）｛returninitTaskExecutor（TtlThreadPoolFactory。DEFAULTCORESIZE，TtlThreadPoolFactory。DEFAULTPOOLSIZE，QUEUECAPACITY）；｝privatestaticThreadPoolTaskExecutorinitTaskExecutor（intcoreSize，intpoolSize，intexecutorQueueCapacity）｛ThreadPoolTaskExecutortaskExecutornewThreadPoolTaskExecutor（）；taskExecutor。setCorePoolSize（coreSize）；taskExecutor。setMaxPoolSize（poolSize）；taskExecutor。setQueueCapacity（executorQueueCapacity）；taskExecutor。setKeepAliveSeconds（120）；taskExecutor。setAllowCoreThreadTimeOut（true）；taskExecutor。setThreadNamePrefix（TaskExecutorttl）；taskExecutor。initialize（）；returntaskE｝privatestaticThreadPoolTaskExecutorgetTtlThreadPoolTaskExecutor（ThreadPoolTaskExecutorexecutor）｛if（nullexecutorexecutorinstanceofThreadPoolTaskExecutorWrapper）｛｝returnnewThreadPoolTaskExecutorWrapper（executor）；｝｝ClassName：LocaleContextHolderDescription：本地化信息上下文holderpublicclassLocalizationContextHolder｛privatestaticTransmittableThreadLocalLocalizationContextlocalizationContextHoldernewTransmittableThreadLocal（）；privatestaticLocalizationInfodefaultLocalizationInfonewLocalizationInfo（）；privateLocalizationContextHolder（）｛｝publicstaticLocalizationContextgetLocalizationContext（）｛returnlocalizationContextHolder。get（）；｝publicstaticvoidresetLocalizationContext（）｛localizationContextHolder。remove（）；｝publicstaticvoidsetLocalizationContext（LocalizationContextlocalizationContext）｛if（localizationContextnull）｛resetLocalizationContext（）；｝else｛localizationContextHolder。set（localizationContext）；｝｝publicstaticvoidsetLocalizationInfo（LocalizationInfolocalizationInfo）｛LocalizationContextlocalizationContextgetLocalizationContext（）；Stringbrand（localizationContextinstanceofBrandLocalizationContext？（（BrandLocalizationContext）localizationContext）。getBrand（）：null）；if（StringUtils。isNotEmpty（brand））｛localizationContextnewSimpleBrandLocalizationContext（localizationInfo，brand）；｝elseif（localizationInfo！null）｛localizationContextnewSimpleLocalizationContext（localizationInfo）；｝else｛localizationC｝setLocalizationContext（localizationContext）；｝publicstaticvoidsetDefaultLocalizationInfo（NullableLocalizationInfolocalizationInfo）｛LocalizationContextHolder。defaultLocalizationInfolocalizationI｝publicstaticLocalizationInfogetLocalizationInfo（）｛LocalizationContextlocalizationContextgetLocalizationContext（）；if（localizationContext！null）｛LocalizationInfolocalizationInfolocalizationContext。getLocalizationInfo（）；if（localizationInfo！null）｛returnlocalizationI｝｝returndefaultLocalizationI｝publicstaticStringgetCountry（）｛returngetLocalizationInfo（）。getCountry（）；｝publicstaticStringgetTimezone（）｛returngetLocalizationInfo（）。getTimezone（）；｝publicstaticStringgetBrand（）｛returngetBrand（getLocalizationContext（））；｝publicstaticStringgetBrand（LocalizationContextlocalizationContext）｛if（localizationContextnull）｛｝if（localizationContextinstanceofBrandLocalizationContext）｛return（（BrandLocalizationContext）localizationContext）。getBrand（）；｝thrownewLocaleException（unsupportedlocalizationContexttype）；｝｝OverridepublicLocaleContextresolveLocaleContext（finalHttpServletRequestrequest）｛parseLocaleCookieIfNecessary（request）；LocaleContextlocaleContextnewTimeZoneAwareLocaleContext（）｛OverridepublicLocalegetLocale（）｛return（Locale）request。getAttribute（LOCALEREQUESTATTRIBUTENAME）；｝OverridepublicTimeZonegetTimeZone（）｛return（TimeZone）request。getAttribute（TIMEZONEREQUESTATTRIBUTENAME）；｝｝；设置线程中的国家标志setLocalizationInfo（request，localeContext。getLocale（））；returnlocaleC｝privatevoidsetLocalizationInfo（HttpServletRequestrequest，Localelocale）｛Stringcountrylocale！null？locale。getCountry（）：Stringlanguagelocale！null？（locale。getLanguage（）locale。getVariant（））：LocaleRequestMessagelocaleRequestMessagelocaleRequestParser。parse（request）；finalStringcountrySfinalStringlanguageSfinalStringbrandStrlocaleRequestMessage。getBrand（）；LocalizationContextHolder。setLocalizationContext（newBrandLocalizationContext（）｛OverridepublicStringgetBrand（）｛returnbrandS｝OverridepublicLocalizationInfogetLocalizationInfo（）｛returnLocalizationInfoAssembler。assemble（countryStr，languageStr）；｝｝）；｝
　　对于定时任务job，因为所有国家都需要执行，所以会把所有国家进行遍历执行，这也可以通过简单的注解和AOP来解决。四、总结
　　本文从业务拓展的角度阐述了在复杂业务场景下如何通过ThreadLocal，过渡到InheritableThreadLocal，再通过TransmittableThreadLocal解决实际业务问题。因为海外的业务在不断的探索中前进，技术也在不断的探索中演进，面对这种复杂多变的情况，我们的应对策略是先做国际化，再做本地化，moreglobal才能morelocal，多国家的隔离只是国际化最基本的起点，未来还有很多业务和技术等着我们去挑战。