Java线程:CPU利用率过高

Question

我正在使用一种服务来读取消息Kafka并将其推入Cassandra.

我正在使用相同的线程架构.

有消息k threads来自Kafka主题.这些写入队列,声明为:

public static BlockingQueue<>

现在有很多线程,比如n写入Cassandra.这是执行此操作的代码:

public void run(){
    LOGGER.log(Level.INFO, "Thread Created: " +Thread.currentThread().getName());
    while (!Thread.currentThread().isInterrupted()) {
        Thread.yield();
        if (!content.isEmpty()) {
            try {
                JSONObject msg = content.remove();
                // JSON
                for(String tableName : tableList){
                    CassandraConnector.getSession().execute(createQuery(tableName, msg));
                }
            } catch (Exception e) {

            }
        }
    }
}

content 是用于读写操作的BlockingQueue.

我在Thread线程的实现中扩展了类,并且有一定数量的线程继续执行,除非被中断.

问题是,这是使用太多的CPU.这是第一行top命令:

  PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND
46232 vishran+  20   0 3010804 188052  14280 S 137.8  3.3   5663:24 java

以下是strace此过程的线程的输出:

strace -t -p 46322
Process 46322 attached
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
15:18:47 sched_yield()                  = 0
....and so on

为什么我使用Thread.yield(),是因为这个

如果您需要任何其他信息进行调试,请告诉我.

现在问题是,如何最大限度地降低CPU利用率？

Answer 1

BlockingQueue的整个目的是它在空时阻塞.因此,消费者线程(填充到Cassandra中的线程)不必手动检查它们是否为空.您可以只调用take(),如果队列为空,则调用将被阻塞,除非它被中断或者有可用的元素.

当一个线程被阻塞时,调度程序可以在其位置安排一些其他线程,这样可以避免调用yield()等等.请记住,只有当优先级大于或等于正在产生的线程的线程可以运行时,yield()才会让位给另一个线程.

public void run(){
    LOGGER.log(Level.INFO, "Thread Created: " +Thread.currentThread().getName());
    try {
            JSONObject msg = content.take();
            // JSON
            for(String tableName : tableList){
                CassandraConnector.getSession().execute(createQuery(tableName, msg));
            }
     } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
     }
}