我们为ELK设置了文件拍子，logstash，elasticsearch和kibana。 我需要在logstash处汇总请求和响应。

我已经按照以下方式配置了管道配置。现在，如果我将单个工作程序用于管道，则日志聚合工作没有任何问题。如果我使用多名工人，则不会发生聚合。有使用多个工作程序和日志聚合的解决方案吗？

 if [transaction] == "request" { 

       aggregate {
                        task_id => "%{id}"
                        code => "
                        map['method'] = event.get('method')
                        map['request'] = event.get('request')
                        map['user'] = event.get('user')
                        map['application'] = event.get('application')"
                        map_action => "create"
         }
                    drop {}#drop the request before persisting,to save indexing space in elasticsearch server
  }
  if [message] =~ "TRANSactION:response" {

         aggregate {
                    task_id => "%{id}"
                    code => "
                    event.set('method',map['method'])
                    event.set('response',map['response'])
                    event.set('user',map['user'])
                    event.set('application',map['application'])"
                    map_action => "update"

         }
}