MapReduce的一些改进

将MapReduce项目中Wait阶段的处理作了一些优化

前文见实现一个MapReduce。

文中我们对有一个地方的处理是这样的：设置一个Wait阶段用于过渡可能产生的分区错误，即处于Wait阶段时worker会过段时间重新申请任务。但这样会产生一些缺点：1等待时间过短会占用大量资源；2等待时间过长任务又会迟迟不受理

但其实这里可以用select语句进行优化，设计得更优雅些

func (m *Master) SendTask(args *RPCArgs, reply *RPCReply) error {

    //mutex the phase
    m.mutex.Lock()
    reply.Phase = m.Phase
    m.mutex.Unlock()

    switch reply.Phase {
    case MapPhase:
        reply.MapTask = <- m.MappersChan
        reply.NReducer = m.NReducer
        //avoid crash
        go m.MapTimeout(reply.MapTask)
    case ReducePhase:
        reply.ReduceTask = <- m.ReducersChan
        go m.ReduceTimeout(reply.ReduceTask)
    //阻塞直到有一个通道下发任务
    //select会随机选择一个未阻塞的通道执行，如果不存在则一直阻塞
    case Wait:
        select {
        case rely.MapTask = <- m.MappersChan:
    	    reply.NReducer = m.NReducer        
            go m.MapTimeout(reply.MapTask)
        case reply.ReduceTask = <- m.ReducersChan:
            go m.ReduceTimeout(reply.ReduceTask)
        }
    }
	
    return nil

}

worker作相应修改即可

 

更进一步想，甚至可以只标记已分发的map任务和完成的map任务，等到所有map任务完成后master才进行shuffle过程，将reduce任务放入通道中。这样，不需要处理的任务就不会进入通道，也就随之阻塞。由select语句和两个通道搭配，甚至不需要设置MapPhase和ReducePhase。笔者碍于时间关系不能重写一遍，希望感兴趣的可以试一试

赏

谢谢你请我吃糖果