这里主要说下在task初始化时的distribute purge相关的类：TrackerDistributedCacheManager
4.6.0的distribute purge行为由4个参数控制。
1
2
3
4
local
.cache.size 
#默认10737418240
mapreduce.tasktracker.cache.
local
.numberdirectories 
#默认10000
mapreduce.tasktracker.cache.
local
.keep.pct 
#默认0.95
mapreduce.tasktracker.distributedcache.checkperiod 
#默认1分钟
在声明一个TrackerDistributedCacheManager类的实例时，会初始化一个CleanupThread类的实例。
1
this
.cleanupThread=newCleanupThread(conf);
并启动
1
2
3
4
5
6
7
8
9
publicvoidstartCleanupThread() {
this
.cleanupThread.start();
  
}
（由TaskTracker类的initialize方法调用
// Initialize DistributedCache
this
.distributedCacheManager=newTrackerDistributedCacheManager(
this
.fConf,taskController);
this
.distributedCacheManager.startCleanupThread();
）
CleanupThread是TrackerDistributedCacheManager类的内部类：
其原理是启动一个thread，定期去触发 BaseDirManager类的checkAndCleanup方法，不会阻塞当前进程。间隔时间由mapreduce.tasktracker.distributedcache.checkperiod控制。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public 
void 
run() {
  
while 
( running) {
    
try 
{
      
Thread. sleep(cleanUpCheckPeriod);
      
baseDirManager.checkAndCleanup();  
//调用checkAndCleanup方法
    
} 
catch 
(IOException e) {
      
LOG.error(
"Exception in DistributedCache CleanupThread." 
, e);
    
} 
catch
(InterruptedException e) {
      
LOG.info(
"Cleanup..." 
,e);
      
//To force us to exit cleanly
      
running = 
false
;
    
} 
catch 
(Throwable t) {
      
exitTaskTracker(t);
    
}
  
}
}
BaseDirManager也是TrackerDistributedCacheManager的内部类，控制distribute cache的删除操作和删除后状态数据的更新：
其中checkAndCleanup方法的实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
    
Collection<CacheStatus> toBeDeletedCache = 
new 
LinkedList<CacheStatus>();
    
HashMap<Path, CacheDir> toBeCleanedBaseDir = 
new 
HashMap<Path, CacheDir>();
.........
         
for 
(Map.Entry<Path, CacheDir> baseDir : properties.entrySet()) { 
//循环遍历mapred.local.dir目录大小和文件量
          
CacheDir baseDirCounts = baseDir.getValue();
          
LOG.debug(baseDir.getKey() + 
": allowedCacheSize=" 
+ allowedCacheSize +
              
",baseDirCounts.size=" 
+ baseDirCounts.size +
              
",allowedCacheSubdirs=" 
+ allowedCacheSubdirs +
              
",baseDirCounts.subdirs=" 
+ baseDirCounts.subdirs);
          
if 
(allowedCacheSize < baseDirCounts.size ||
              
allowedCacheSubdirs < baseDirCounts.subdirs) { 
//触发purge的条件（local.cache.size小于某一个目录大小，mapreduce.tasktracker.cache.local.numberdirectories小于某一个文件下的文件数量）
            
CacheDir tcc = 
new 
CacheDir();
            
tcc.size = baseDirCounts.size - allowedCacheSizeCleanupGoal;
            
tcc.subdirs = baseDirCounts.subdirs - allowedCacheSubdirsCleanupGoal;
            
toBeCleanedBaseDir.put(baseDir.getKey(), tcc); 
//生成需要删除的目录的HashMap
          
}
        
}
实际的删除动作：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// do the deletion, after releasing the global lock
for 
(CacheStatus cacheStatus : toBeDeletedCache) {  
//循环遍历需要删除的cache
  
cacheStatus. lock.lock();  
//获取删除对象的锁
  
try 
{
    
Path localizedDir = cacheStatus.getLocalizedUniqueDir();
    
if 
(cacheStatus.user == 
null
) {
      
TrackerDistributedCacheManager. LOG.info(
"Deleted path " 
+ localizedDir);
      
try 
{
        
localFs.delete(localizedDir, 
true
);  
//public的情况调用FileSystem的delete方法
      
} 
catch 
(IOException e) {
        
TrackerDistributedCacheManager. LOG.warn(
"Could not delete distributed cache empty directory "
                 
+ localizedDir, e);
      
}
    
} 
else 
{ 
      
TrackerDistributedCacheManager. LOG.info(
"Deleted path " 
+ localizedDir + 
" as " 
+ cacheStatus.user );
      
String base = cacheStatus.getBaseDir().toString();
      
String userDir = TaskTracker.getUserDir(cacheStatus. user);
      
int 
skip = base.length() + 
1 
+ userDir.length() + 
1
;
      
String relative = localizedDir.toString().substring(skip);
      
taskController.deleteAsUser(cacheStatus.user , relative);  
//private的情况调用TaskController的deleteAsUser的方法
    
}
    
deleteCacheInfoUpdate(cacheStatus);
  
} 
finally 
{
    
cacheStatus. lock.unlock();
  
}
}