groovy 脚本中的多线程答案

【问题标题】：Multithreading in groovy scriptgroovy 脚本中的多线程
【发布时间】：2022-11-10 12:11:15
【问题描述】：

我有一个生产者消费者问题，我有单个生产者推入阻塞队列和单个消费者从队列中消费。一旦使用了一条消息，我就会对该批消息进行多项操作。如何并行处理每批消息的逻辑处理。下面是代码sn-p。还建议我是否应该考虑多个消费者来完成这项任务。

ThreadX = Thread.start('producer') {
//data retrieve from DB
while(row){
   queue.put(message)
  }
   queue.put("KILL")
}
ThreadY = Thread.start('Consumer') {
    while(true){
        sleep(200)
      //  print(Thread.currentThread().name)
        def jsonSlurper = new JsonSlurper()
        def var = jsonSlurper.parseText(queue.take().toString())
        if(var.getAt(0).equals("KILL"))
            return
        var.each { fileExists(it) } // **need  parallelize this part**
    }

boolean fileExists(key){
    if(key) {
    //some logic
        sleep 1000
    }
}
}

更新：尝试了以下代码，但它以某种方式只处理消费者消费的第一批 10 条消息

ExecutorService exeSvc = Executors.newFixedThreadPool(5)

ThreadY = Thread.start('Consumer') {
    while(true){
        sleep(200)
      //  print(Thread.currentThread().name)
        def jsonSlurper = new JsonSlurper()
        def var = jsonSlurper.parseText(queue.take().toString())
        if(var.getAt(0).equals("KILL"))
            return
        var.each { exeSvc.execute({-> fileExists(it)
            sleep(200)
        }) }
    }
}

请帮忙

【问题讨论】：

标签： multithreading groovy parallel-processing

【解决方案1】：

var.each { exeSvc.execute({-> fileExists(it) 行中有一个错误，其中外部闭包中的隐式变量it 在内部闭包中使用。应该类似于var.each { fileName -> exeSvc.execute { fileExists(filName)。除此之外，我只是添加了一些日志用于故障排除和确认执行流程。以下脚本的工作版本：

import groovy.json.JsonSlurper

import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

Queue queue = new LinkedBlockingQueue<String>()
def messages = [
        '["key1", "key2", "key3", "key4", "key5"]',
        '["key6", "key7", "key8", "key9", "key10"]',
        '["key11", "key12", "key13", "key14", "key15"]',
        '["key16", "key17", "key18", "key19", "key20"]',
        '["key21", "key22", "key23", "key24", "key25"]',
]

Thread.start('producer') {
    messages.each { message ->
        println("Producing message $message")
        queue.put(message)
    }
    queue.put('KILL')
}

ExecutorService exeSvc = Executors.newFixedThreadPool(5)
def jsonSlurper = new JsonSlurper()

Thread.start('Consumer') {
    while (true) {
        sleep(200)

        String message = queue.take()
        println("Consumed message $message")
        if (message == 'KILL') {
            exeSvc.shutdown()
            exeSvc.awaitTermination(10, TimeUnit.SECONDS)
            return
        }

        def var = jsonSlurper.parseText(message)
        var.each { fileName ->
            exeSvc.execute {
                fileExists(fileName)
                sleep(200)
            }
        }
    }
}

boolean fileExists(key) {
    println("Key: $key")
    if (key) {
        sleep 1000
    }
}

【讨论】：