Goroutines 및 메시지 중복 제거

그래서 이벤트 대기열과 무한 루프의 해당 대기열에서 이벤트를 가져 오는 일부 goroutines을 처리하여 채널로 결과를 보냅니다. 다른 대기열이 동일한 이벤트를 제공 할 수 있으므로 각 이벤트가 정확히 한 번 채널에 전송되도록해야하며 다른 대기열에있는 해당 메시지의 발생이 무시됩니다. 나는 그것이 건축상의 문제라고 생각한다. 그러나 나는 이것을 올바르게 다루는 방법을 알 수 없다.Goroutines 및 메시지 중복 제거

내 현재 코드의 단순화 된 버전은 아래에 있습니다. 얻을 들어오는 이벤트를 처리

Goroutines은 다음과 같이 다소 찾습니다

이

func (q *Queue) ProcessEvents(handler Handler) { 
    lastEvent = 0 
    for { 
     events = getEvents(lastEvent) 
     for _, e := range events { 
      if e.ID > lastEvent { 
       lastEvent = event.ID 
      } 
      handler.Handle(e) 
     } 
    } 
}

처리기 :

type Handler struct { 
    c chan Event 
} 

func (h *Handler) Handle(event *Event) { 
    //event processing omitted 
    h.c <- event //Now it just sends a processed event into the channel no matter what. 
}

그리고 주에서

() 나는

func main() { 
    msgc := make(chan Event) 
    for _, q := range queues { 
     go func(queue Queue) { 
      queue.ProcessEvents(&Handler{msgc}) 
     } 
    } 
}

출처

2017-11-15 No Matter

채널의 수신 측에서이를 수행하고 동일한 이벤트를 두 번 이상 처리하지 않으려는 것처럼 보입니다. –

그러나 atm 루틴이 서로에 대해 알지 못한다면 이벤트가 이미 다른 goroutine에 의해 처리되었는지 어떻게 확인합니까? 그것은 정말로 중요한 질문입니다. –

단일 스레드 인 경우 어떻게합니까? 이것은 goroutines에 관한 것이 아닙니다. 중복 제거를 원할 경우 메시지가 복제본인지 여부를 알아야합니다. 즉, 당신이 얻은 모든 메시지를 (잠재적으로) 저장하고 모든 새로운 메시지를 모든 이전 메시지와 비교하여 이전에 본 적이 있는지 확인하십시오. – Adrian

그래서 당신이 대표 마십시오 현재 아키텍처는 다음과 같습니다 :

이러한 유형의 솔루션을 사용하면 생성자는 이벤트가 이미 방출되었는지 확인하기 위해 공유 리소스를 검사해야합니다. 이것은 다음과 같이 보일 수 있습니다이도 경합과 함께 최상의 시나리오에서 임계 영역에서 수행되는 작업의 소량을 고려 큰 오버 헤드까지입니다 잠금/잠금 해제를 요구

var hasEmmited map[string]bool 
var lock sync.Mutex 

func HasEmitted(event e) bool { 
    lock.Lock() 
    defer lock.Unlock() 
    e,ok := hasEmmited[e.ID] 
    return e && ok 
} 

func SetEmmited(event e) { 
    lock.Lock() 
    defer lock.Lock() 
    hasEmmited[e.ID] = true 
}

두 번째 다이어그램과 같이 아키텍처가 약간 변경되었으므로 한 번의 작업으로 모든 잠금을 수행하지 않고도 필터링을 수행 할 수 있습니다.