聊聊rocketmqstreams的ILeaseServic

　　序
　　本文主要研究一下rocketmqstreams的ILeaseServiceILeaseService通过db实现租约和锁，可以更轻量级，减少其他中间件的依赖使用主备场景，只有一个实例运行，当当前实例挂掉，在一定时间内，会被其他实例接手也可以用于全局锁publicinterfaceILeaseService｛默认锁定时间staticfinalintDEFALUTLOCKTIME605；检查某用户当前时间是否具有租约。这个方法是纯内存操作，无性能开销returntrue，租约有效；false，租约无效booleanhasLease（Stringname）；申请租约，会启动一个线程，不停申请租约，直到申请成功。申请成功后，每租期2续约。如果目前被其他租户获取租约，只有在对方租约失效，后才允许新的租户获取租约paramname租约名称，无特殊要求，相同名称会竞争租约voidstartLeaseTask（Stringname）；申请租约，会启动一个线程，不停申请租约，直到申请成功。申请成功后，每租期2续约。如果目前被其他租户获取租约，只有在对方租约失效，后才允许新的租户获取租约paramname租约名称，无特殊要求，相同名称会竞争租约paramcallback当第一获取租约时，回调此函数voidstartLeaseTask（finalStringname，ILeaseGetCallbackcallback）；申请租约，会启动一个线程，不停申请租约，直到申请成功。申请成功后，每租期2续约。如果目前被其他租户获取租约，只有在对方租约失效，后才允许新的租户获取租约paramname租约名称，无特殊要求，相同名称会竞争租约paramleaseTermSecond租期，在租期内可以做业务处理，单位是秒paramcallback当第一获取租约时，回调此函数voidstartLeaseTask（finalStringname，intleaseTermSecond，ILeaseGetCallbackcallback）；申请锁，无论成功与否，立刻返回。如果不释放，最大锁定时间是5分钟paramname业务名称paramlockerName锁名称return是否枷锁成功booleanlock（Stringname，StringlockerName）；申请锁，无论成功与否，立刻返回。默认锁定时间是5分钟paramname业务名称paramlockerName锁名称paramlockTimeSecond如果不释放，锁定的最大时间，单位是秒return是否枷锁成功returnbooleanlock（Stringname，StringlockerName，intlockTimeSecond）；申请锁，如果没有则等待，等待时间可以指定，如果是1则无限等待。如果不释放，最大锁定时间是5分钟paramname业务名称paramlockerName锁名称paramwaitTime没获取锁时，最大等待多长时间，如果是1则无限等待return是否枷锁成功booleantryLocker（Stringname，StringlockerName，longwaitTime）；申请锁，如果没有则等待，等待时间可以指定，如果是1则无限等待。如果不释放，最大锁定时间是lockTimeSecondparamname业务名称paramlockerName锁名称paramwaitTime没获取锁时，最大等待多长时间，如果是1则无限等待paramlockTimeSecond如果不释放，锁定的最大时间，单位是秒return是否枷锁成功booleantryLocker（Stringname，StringlockerName，longwaitTime，intlockTimeSecond）；释放锁paramnameparamlockerNamereturnbooleanunlock（Stringname，StringlockerName）；对于已经获取锁的，可以通过这个方法，一直持有锁。和租约的区别是，当释放锁后，无其他实例抢占。无法实现主备模式paramname业务名称paramlockerName锁名称paramlockTimeSecond租期，这个方法会自动续约，如果不主动释放，会一直持有锁return是否成功获取锁booleanholdLock（Stringname，StringlockerName，intlockTimeSecond）；是否持有锁，不会申请锁。如果以前申请过，且未过期，返回true，否则返回falseparamname业务名称paramlockerName锁名称returnbooleanhasHoldLock（Stringname，StringlockerName）；ListLeaseInfoqueryLockedInstanceByNamePrefix（Stringname，StringlockerNamePrefix）；｝ILeaseService接口定义了hasLease、startLeaseTask、lock、tryLocker、unlock、holdLock、hasHoldLock、queryLockedInstanceByNamePrefix方法BasedLesaseImplpublicabstractclassBasedLesaseImplimplementsILeaseService｛privatestaticfinalLogLOGLogFactory。getLog（BasedLesaseImpl。class）；privatestaticfinalStringCONSISTENTHASHPREFIXconsistenthash；privatestaticfinalAtomicBooleansyncStartnewAtomicBoolean（false）；privatestaticfinalintsynTime120；5分钟的一致性hash同步时间太久了，改为2分钟protectedScheduledExecutorServicetaskExecutornull；protectedintleaseTerm3002；租约时间protectedtransientJDBCDriverjdbcDataSourcenull；protectedILeaseStorageleaseStorage；protectedvolatileMapString，DateleaseName2DatenewConcurrentHashMap（）；每个leasename对应的租约到期时间publicBasedLesaseImpl（）｛taskExecutornewScheduledThreadPoolExecutor（10）；｝leasename：consistenthaship，leaseuserip：ip，定时刷新leaseinfo表，检查一致性hash环的节点情况paramnamereturnOverridepublicbooleanhasLease（Stringname）｛内存中没有租约信息则表示没有租约DateleaseEndTimeleaseName2Date。get（name）；if（leaseEndTimenull）｛LOG。info（内存中根据name没有查询到租约信息，表示没有租约）；returnfalse；｝LOG。info（查询是否有租约name：name，当前时间：newSimpleDateFormat（yyyyMMddHH：mm：ss）。format（newDate（））租约到期时间newSimpleDateFormat（yyyyMMddHH：mm：ss）。format（leaseEndTime））；有租约时间，并且租约时间大于当前时间，表示有租约信息if（newDate（）。before（leaseEndTime））｛returntrue；｝returnfalse；｝privatefinalMapString，AtomicBooleanstartLeaseMapnewHashMap（）；OverridepublicvoidstartLeaseTask（finalStringname）｛startLeaseTask（name，this。leaseTerm，null）；｝OverridepublicvoidstartLeaseTask（finalStringname，ILeaseGetCallbackcallback）｛startLeaseTask（name，this。leaseTerm，callback）；｝OverridepublicvoidstartLeaseTask（finalStringname，intleaseTerm，ILeaseGetCallbackcallback）｛ApplyTaskapplyTasknewApplyTask（leaseTerm，name，callback）；startLeaseTask（name，applyTask，leaseTerm2，true）；｝启动定时器，定时执行任务，确保任务可重入paramnameparamrunnable具体任务paramscheduleTime调度时间paramstartNow是否立刻启动一次protectedvoidstartLeaseTask（finalStringname，Runnablerunnable，intscheduleTime，booleanstartNow）｛AtomicBooleanisStartLeasestartLeaseMap。get（name）；多次调用，只启动一次定时任务if（isStartLeasenull）｛synchronized（this）｛isStartLeasestartLeaseMap。get（name）；if（isStartLeasenull）｛isStartLeasenewAtomicBoolean（false）；startLeaseMap。put（name，isStartLease）；｝｝｝if（isStartLease。compareAndSet（false，true））｛if（startNow）｛runnable。run（）；｝taskExecutor。scheduleWithFixedDelay（runnable，0，scheduleTime，TimeUnit。SECONDS）；｝｝。。。。。。｝BasedLesaseImpl声明实现了ILeaseService，它依赖ILeaseStorage，startLeaseTask方法会创建ApplyTask，然后以固定间隔调度执行ApplyTask续约任务protectedclassApplyTaskimplementsRunnable｛protectedStringname；protectedintleaseTerm；protectedILeaseGetCallbackcallback；publicApplyTask（intleaseTerm，Stringname）｛this（leaseTerm，name，null）；｝publicApplyTask（intleaseTerm，Stringname，ILeaseGetCallbackcallback）｛this。namename；this。leaseTermleaseTerm；this。callbackcallback；｝Overridepublicvoidrun（）｛try｛LOG。info（LeaseServiceImplname：name开始获取租约。。。）；AtomicBooleannewApplyLeasenewAtomicBoolean（false）；DateleaseDateapplyLeaseTask（leaseTerm，name，newApplyLease）；if（leaseDate！null）｛leaseName2Date。put（name，leaseDate）；LOG。info（LeaseServiceImpl，name：namegetSelfUser（）获取租约成功，租约到期时间为newSimpleDateFormat（yyyyMMddHH：mm：ss）。format（leaseDate））；｝else｛fix。2020。08。13这时name对应的租约可能还在有效期内，或者本机还持有租约，需要removeleaseName2Date。remove（name）；LOG。info（LeaseServiceImplname：namegetSelfUser（）获取租约失败）；｝if（newApplyLease。get（）callback！null）｛callback。callback（leaseDate）；｝｝catch（Exceptione）｛LOG。error（LeaseServiceImplname：namegetSelfUser（）获取租约出现异常，e）；｝｝｝申请租约，如果当期租约有效，直接更新一个租约周期，如果当前租约无效，先查询是否有有效的租约，如果有申请失败，否则直接申请租约protectedDateapplyLeaseTask（intleaseTerm，Stringname，AtomicBooleannewApplyLease）｛计算下一次租约时间当前时间租约时长DatenextLeaseDateDateUtil。addSecond（newDate（），leaseTerm）；1如果已经有租约，则更新租约时间（内存和数据库）即可if（hasLease（name））｛LOG。info（用户已有租约，更新数据库和内存中的租约信息）；更新数据库LeaseInfoleaseInfoqueryValidateLease（name）；if（leaseInfonull）｛LOG。error（LeaseServiceImplapplyLeaseTaskleaseInfoisnull）；returnnull；｝fix。2020。08。13，与本机ip相等且满足一致性hash分配策略，才续约，其他情况为nullStringleaseUserIpleaseInfo。getLeaseUserIp（）；if（！leaseUserIp。equals（getSelfUser（）））｛returnnull；｝leaseInfo。setLeaseEndDate（nextLeaseDate）；updateLeaseInfo（leaseInfo）；returnnextLeaseDate；｝2没有租约情况判断是否可以获取租约，只要租约没有被其他人获取，则说明有有效租约booleansuccesscanGetLease（name）；if（！success）｛表示被其他机器获取到了有效的租约LOG。info（其他机器获取到了有效的租约）；returnnull；｝3没有租约而且可以获取租约的情况，则尝试使用数据库原子更新的方式获取租约，保证只有一台机器成功获取租约，而且可以运行booleanflagtryGetLease（name，nextLeaseDate）；if（flag）｛获取租约成功newApplyLease。set（true）；returnnextLeaseDate；｝returnnull；｝ApplyTask内部调用applyLeaseTask，如果已有租约则更新租约时间，没有租约则判断是否可以获取租约，可以则执行tryGetLeasetryGetLease更新数据库，占用租期并更新租期时间paramtimeprotectedbooleantryGetLease（Stringname，Datetime）｛LOG。info（尝试获取租约leasenameis：name下次到期时间：newSimpleDateFormat（yyyyMMddHH：mm：ss）。format（time））；LeaseInfovalidateLeaseInfoqueryValidateLease（name）；if（validateLeaseInfonull）｛这里有两种情况1数据库里面没有租约信息2数据库里面有租约信息但是已经过期IntegercountcountLeaseInfo（name）；if（countnullcount0）｛表示现在数据库里面没有任何租约信息，插入租约成功则表示获取成功，失败表示在这一时刻其他机器获取了租约LOG。info（数据库中暂时没有租约信息，尝试原子插入租约：name）；fix。2020。08。13，经过一致性hash计算，该名字的任务不应该在本机执行，直接返回，无需插入。只有分配到hash执行权限的机器才可以插入并获取租约if（！getSelfUser（）。equals（getConsistentHashHost（name）））｛returnfalse；｝validateLeaseInfonewLeaseInfo（）；validateLeaseInfo。setLeaseName（name）；validateLeaseInfo。setLeaseUserIp（getSelfUser（））；validateLeaseInfo。setLeaseEndDate（time）；validateLeaseInfo。setStatus（1）；validateLeaseInfo。setVersion（1）；if（insert（validateLeaseInfo））｛LOG。info（数据库中暂时没有租约信息，原子插入成功，获取租约成功：name）；returntrue；｝else｛LOG。info（数据库中暂时没有租约信息，原子插入失败，已经被其他机器获取租约：name）；returnfalse；｝｝else｛表示数据库里面有一条但是无效，这里需要两台机器按照version进行原子更新，更新成功的获取租约LOG。info（数据库中有一条无效的租约信息，尝试根据版本号去原子更新租约信息：name）；LeaseInfoinValidateLeaseInfoqueryInValidateLease（name）；if（inValidateLeaseInfonull）｛说明这个时候另外一台机器获取成功了LOG。info（另外一台机器获取成功了租约：name）；returnfalse；｝fix。2020。08。13，机器重启之后，该名字的任务已经不分配在此机器上执行，直接返回，无需更新数据库if（！getSelfUser（）。equals（getConsistentHashHost（name）））｛returnfalse；｝inValidateLeaseInfo。setLeaseName（name）；inValidateLeaseInfo。setLeaseUserIp（getSelfUser（））；inValidateLeaseInfo。setLeaseEndDate（time）；inValidateLeaseInfo。setStatus（1）；booleansuccessupdateDBLeaseInfo（inValidateLeaseInfo）；if（success）｛LOG。info（LeaseServiceImpl原子更新租约成功，当前机器获取到了租约信息：name）；｝else｛LOG。info（LeaseServiceImpl原子更新租约失败，租约被其他机器获取：name）；｝returnsuccess；｝｝else｛判断是否是自己获取了租约，如果是自己获取了租约则更新时间（内存和数据库），这里是为了解决机器重启的情况，机器重启，内存中没有租约信息，但是实际上该用户是有租约权限的fix。2020。08。13，租约的ip与本机ip相等，且满足一致性hash策略，才会被本机执行StringleaseUserIpvalidateLeaseInfo。getLeaseUserIp（）；if（leaseUserIp。equals（getSelfUser（）））｛如果当期用户有租约信息，则更新数据库validateLeaseInfo。setLeaseEndDate（time）；booleanhasUpdateupdateLeaseInfo（validateLeaseInfo）；if（hasUpdate）｛LOG。info（LeaseServiceImpl机器重启情况，当前用户有租约信息，并且更新数据库成功，租约信息为name：validateLeaseInfo。getLeaseName（）ip：validateLeaseInfo。getLeaseUserIp（）到期时间：newSimpleDateFormat（yyyyMMddHH：mm：ss）。format（validateLeaseInfo。getLeaseEndDate（）））；returntrue；｝else｛LOG。info（LeaseServiceImpl机器重启情况，当前用户有租约信息，并且更新数据库失败，表示失去租约：name）；returnfalse；｝｝LOG。info（LeaseServiceImpl租约被其他机器获取，租约信息为name：validateLeaseInfo。getLeaseName（）ip：validateLeaseInfo。getLeaseUserIp（）到期时间：newSimpleDateFormat（yyyyMMddHH：mm：ss）。format（validateLeaseInfo。getLeaseEndDate（）））；returnfalse；｝｝protectedLeaseInfoqueryValidateLease（Stringname）｛StringsqlSELECTFROMleaseinfoWHEREleasenamenameandstatus1andleaseendtimenow（）；LOG。info（LeaseServiceImplqueryvalidateleasesql：sql）；returnqueryLease（name，sql）；returnleaseStorage。queryValidateLease（name）；｝tryGetLease先通过queryValidateLease查询租约信息，若没有租约则插入，若过期则根据版本号更新，若已有租约则判断是否是自己获取了租约，是则更新租约信息LeaseServiceImplpublicclassLeaseServiceImplextendsBasedLesaseImpl｛privatestaticfinalLogLOGLogFactory。getLog（LeaseServiceImpl。class）；privatetransientConcurrentHashMapString，HoldLockTaskholdLockTasksnewConcurrentHashMap（）；protectedConcurrentHashMapString，HoldLockFuntureseizeLockingFunturesnewConcurrentHashMap（）；如果是抢占锁状态中，则不允许申请锁publicLeaseServiceImpl（）｛super（）；｝尝试获取锁，可以等待waitTime，如果到点未返回，则直接返回。如果是1，则一直等待paramname业务名称paramlockerName锁名称paramwaitTime等待时间，是微秒单位returnOverridepublicbooleantryLocker（Stringname，StringlockerName，longwaitTime）｛returntryLocker（name，lockerName，waitTime，ILeaseService。DEFALUTLOCKTIME）；｝OverridepublicbooleantryLocker（Stringname，StringlockerName，longwaitTime，intlockTimeSecond）｛longnowSystem。currentTimeMillis（）；booleansuccesslock（name，lockerName，lockTimeSecond）；while（！success）｛if（waitTime1（System。currentTimeMillis（）nowwaitTime））｛break；｝successlock（name，lockerName，lockTimeSecond）；if（success）｛returnsuccess；｝try｛Thread。sleep（100）；｝catch（InterruptedExceptione）｛LOG。error（LeaseServiceImpltrylockererror，e）；｝｝returnsuccess；｝Overridepublicbooleanlock（Stringname，StringlockerName）｛returnlock（name，lockerName，ILeaseService。DEFALUTLOCKTIME）；｝Overridepublicbooleanlock（Stringname，StringlockerName，intleaseSecond）｛lockerNamecreateLockName（name，lockerName）；FuturefutureseizeLockingFuntures。get（lockerName）；if（future！null（（HoldLockFunture）future）。isDonefalse）｛returnfalse；｝DatenextLeaseDateDateUtil。addSecond（newDate（），leaseSecond）；默认锁定5分钟，用完需要立刻释放。如果时间不同步，可能导致锁失败returntryGetLease（lockerName，nextLeaseDate）；｝Overridepublicbooleanunlock（Stringname，StringlockerName）｛LOG。info（LeaseServiceImplunlock，name：name）；lockerNamecreateLockName（name，lockerName）；LeaseInfovalidateLeaseInfoqueryValidateLease（lockerName）；if（validateLeaseInfonull）｛LOG。warn（LeaseServiceImplunlock，validateLeaseInfoisnull，lockerName：lockerName）；｝if（validateLeaseInfo！nullvalidateLeaseInfo。getLeaseUserIp（）。equals（getSelfUser（）））｛validateLeaseInfo。setStatus（0）；updateDBLeaseInfo（validateLeaseInfo）；｝HoldLockTaskholdLockTaskholdLockTasks。remove（lockerName）；if（holdLockTask！null）｛holdLockTask。close（）；｝leaseName2Date。remove（lockerName）；returnfalse；｝如果有锁，则一直持有，如果不能获取，则结束。和租约不同，租约是没有也会尝试重试，一备对方挂机，自己可以接手工作paramnameparamsecondeNameparamlockTimeSecond获取锁的时间returnOverridepublicbooleanholdLock（Stringname，StringsecondeName，intlockTimeSecond）｛if（hasHoldLock（name，secondeName））｛returntrue；｝synchronized（this）｛if（hasHoldLock（name，secondeName））｛returntrue；｝StringlockerNamecreateLockName（name，secondeName）；DatenextLeaseDateDateUtil。addSecond（newDate（），lockTimeSecond）；booleansuccesstryGetLease（lockerName，nextLeaseDate）；申请锁，锁的时间是leaseTermif（！success）｛returnfalse；｝leaseName2Date。put（lockerName，nextLeaseDate）；if（！holdLockTasks。containsKey（lockerName））｛HoldLockTaskholdLockTasknewHoldLockTask（lockTimeSecond，lockerName，this）；holdLockTask。start（）；holdLockTasks。putIfAbsent（lockerName，holdLockTask）；｝｝returntrue；｝是否持有锁，不访问数据库，直接看本地paramnameparamsecondeNamereturnOverridepublicbooleanhasHoldLock（Stringname，StringsecondeName）｛StringlockerNamecreateLockName（name，secondeName）；returnhasLease（lockerName）；｝OverridepublicListLeaseInfoqueryLockedInstanceByNamePrefix（Stringname，StringlockerNamePrefix）｛StringleaseNamePrefixMapKeyUtil。createKey（name，lockerNamePrefix）；returnqueryValidateLeaseByNamePrefix（leaseNamePrefix）；｝。。。。。。｝LeaseServiceImpl继承了BasedLesaseImpl，tryLocker方法会根据等待时间循环执行lock，lock方法则执行tryGetLease，unlock方法则更新租约信息，同时移除内存记录；holdLock则通过hasHoldLock判断是否持有锁，若有则返回，没有则执行tryGetLeaseILeaseStoragepublicinterfaceILeaseStorage｛更新leaseinfo，需要是原子操作，存储保障多线程操作的原子性paramleaseInfo租约表数据returnbooleanupdateLeaseInfo（LeaseInfoleaseInfo）；统计这个租约名称下，LeaseInfo对象个数paramleaseName租约名称，无特殊要求，相同名称会竞争租约returnIntegercountLeaseInfo（StringleaseName）；查询无效的的租约paramleaseName租约名称，无特殊要求，相同名称会竞争租约returnLeaseInfoqueryInValidateLease（StringleaseName）；查询有效的的租约paramleaseName租约名称，无特殊要求，相同名称会竞争租约returnLeaseInfoqueryValidateLease（StringleaseName）；按前缀查询有效的租约信息paramnamePrefixreturnListLeaseInfoqueryValidateLeaseByNamePrefix（StringnamePrefix）；增加租约paramleaseInfo租约名称，无特殊要求，相同名称会竞争租约voidaddLeaseInfo（LeaseInfoleaseInfo）；｝ILeaseStorage接口定义了updateLeaseInfo、countLeaseInfo、queryInValidateLease、queryValidateLease、queryValidateLeaseByNamePrefix、addLeaseInfo方法DBLeaseStoragepublicclassDBLeaseStorageimplementsILeaseStorage｛privatestaticfinalLogLOGLogFactory。getLog（DBLeaseStorage。class）；protectedJDBCDriverjdbcDataSource；privateStringurl；protectedStringuserName；protectedStringpassword；protectedStringjdbc；publicDBLeaseStorage（Stringjdbc，Stringurl，StringuserName，Stringpassword）｛this。jdbcjdbc；this。urlurl；this。userNameuserName；this。passwordpassword；jdbcDataSourceDriverBuilder。createDriver（jdbc，url，userName，password）；｝OverridepublicbooleanupdateLeaseInfo（LeaseInfoleaseInfo）｛StringsqlUPDATEleaseinfoSETversionversion1，status｛status｝，gmtmodifiednow（）；StringwhereSQLWHEREid｛id｝andversion｛version｝；if（StringUtil。isNotEmpty（leaseInfo。getLeaseName（）））｛sql，leasename｛leaseName｝；｝if（StringUtil。isNotEmpty（leaseInfo。getLeaseUserIp（）））｛sql，leaseuserip｛leaseUserIp｝；｝if（leaseInfo。getLeaseEndDate（）！null）｛sql，leaseendtime｛leaseEndDate｝；｝sqlwhereSQL；sqlSQLUtil。parseIbatisSQL（leaseInfo，sql）；try｛intcountgetOrCreateJDBCDataSource（）。update（sql）；booleansuccesscount0；if（success）｛synchronized（this）｛leaseInfo。setVersion（leaseInfo。getVersion（）1）；｝｝else｛System。out。println（count）；｝returnsuccess；｝catch（Exceptione）｛LOG。error（LeaseServiceImplupdateLeaseInfoexcuteUpdateerror，e）；thrownewRuntimeException（executesqlerrorsql，e）；｝｝OverridepublicIntegercountLeaseInfo（StringleaseName）｛StringsqlSELECTcount（）ascFROMleaseinfoWHEREleasenameleaseNameandstatus1；try｛ListMapString，ObjectrowsgetOrCreateJDBCDataSource（）。queryForList（sql）；if（rowsnullrows。size（）0）｛returnnull；｝Longvalue（Long）rows。get（0）。get（c）；returnvalue。intValue（）；｝catch（Exceptione）｛thrownewRuntimeException（executesqlerrorsql，e）；｝｝OverridepublicLeaseInfoqueryInValidateLease（StringleaseName）｛StringsqlSELECTFROMleaseinfoWHEREleasenameleaseNameandstatus1andleaseendtimeDateUtil。getCurrentTimeString（）；LOG。info（LeaseServiceImplqueryInValidateLeasebuilder：sql）；returnqueryLease（leaseName，sql）；｝OverridepublicLeaseInfoqueryValidateLease（StringleaseName）｛StringsqlSELECTFROMleaseinfoWHEREleasenameleaseNameandstatus1andleaseendtimenow（）；returnqueryLease（leaseName，sql）；｝OverridepublicListLeaseInfoqueryValidateLeaseByNamePrefix（StringnamePrefix）｛StringsqlSELECTFROMleaseinfoWHEREleasenamelikenamePrefixandstatus1andleaseendtimenow（）；try｛ListLeaseInfoleaseInfosnewArrayList（）；ListMapString，ObjectrowsgetOrCreateJDBCDataSource（）。queryForList（sql）；if（rowsnullrows。size（）0）｛returnnull；｝for（MapString，Objectrow：rows）｛LeaseInfoleaseInfoconvert（row）；leaseInfos。add（leaseInfo）；｝returnleaseInfos；｝catch（Exceptione）｛thrownewRuntimeException（executesqlerrorsql，e）；｝｝OverridepublicvoidaddLeaseInfo（LeaseInfoleaseInfo）｛StringsqlREPLACEINTOleaseinfo（leasename，leaseuserip，leaseendtime，status，version，gmtcreate，gmtmodified）VALUES（｛leaseName｝，｛leaseUserIp｝，｛leaseEndDate｝，｛status｝，｛version｝，now（），now（））；sqlSQLUtil。parseIbatisSQL（leaseInfo，sql）；try｛getOrCreateJDBCDataSource（）。execute（sql）；｝catch（Exceptione）｛LOG。error（LeaseServiceImplexecutesqlerror，sql：sql，e）；thrownewRuntimeException（executesqlerrorsql，e）；｝｝protectedJDBCDrivergetOrCreateJDBCDataSource（）｛if（this。jdbcDataSourcenull！this。jdbcDataSource。isValidate（））｛synchronized（this）｛if（this。jdbcDataSourcenull！this。jdbcDataSource。isValidate（））｛this。jdbcDataSourceDriverBuilder。createDriver（this。jdbc，this。url，this。userName，this。password）；｝｝｝returnjdbcDataSource；｝protectedLeaseInfoqueryLease（Stringname，Stringsql）｛try｛ListMapString，ObjectrowsgetOrCreateJDBCDataSource（）。queryForList（sql）；if（rowsnullrows。size（）0）｛returnnull；｝returnconvert（rows。get（0））；｝catch（Exceptione）｛thrownewRuntimeException（executesqlerrorsql，e）；｝｝protectedLeaseInfoconvert（MapString，Objectmap）｛LeaseInfoleaseInfonewLeaseInfo（）；leaseInfo。setId（getMapLongValue（id，map））；leaseInfo。setCreateTime（getMapDateValue（gmtcreate，map））；leaseInfo。setLeaseEndDate（getMapDateValue（leaseendtime，map））；leaseInfo。setLeaseName（getMapValue（leasename，map，String。class））；leaseInfo。setLeaseUserIp（getMapValue（leaseuserip，map，String。class））；IntegerstatusgetMapValue（status，map，Integer。class）；if（status！null）｛leaseInfo。setStatus（status）；｝leaseInfo。setUpdateTime（getMapDateValue（gmtmodified，map））；LongversiongetMapLongValue（version，map）；if（version！null）｛leaseInfo。setVersion（version）；｝returnleaseInfo；｝SuppressWarnings（unchecked）privateTTgetMapValue（StringfieldName，MapString，Objectmap，ClassTintegerClass）｛Objectvaluemap。get（fieldName）；if（valuenull）｛returnnull；｝return（T）value；｝privateLonggetMapLongValue（StringfieldName，MapString，Objectmap）｛Objectvaluemap。get（fieldName）；if（valuenull）｛returnnull；｝if（valueinstanceofLong）｛return（Long）value；｝if（valueinstanceofBigInteger）｛return（（BigInteger）value）。longValue（）；｝returnnull；｝privateDategetMapDateValue（StringfieldName，MapString，Objectmap）｛Objectvaluemap。get（fieldName）；if（valuenull）｛returnnull；｝if（valueinstanceofDate）｛return（Date）value；｝if（valueinstanceofString）｛returnDateUtil。parseTime（（（String）value））；｝returnnull；｝｝DBLeaseStorage实现了ILeaseStorage接口，使用jdbc实现了其方法小结
　　rocketmqstreams的LeaseService基于db实现了租约和锁，可用于主备场景切换。