使用线程的Java生产者使用者永不终止 [英] Producer Consumer in Java using threads never terminates

问题描述

我有一个用Java实现的生产者－消费者问题，我希望生产者线程在特定时间段内运行，例如1天后，将对象放入BlockingQueue-特别是推文中，并通过Twitter4j从Twitter Streaming API流式传输，并使用消费者线程从队列中消费这些对象并将其写入文件.我已经使用了

I have a Producer-Consumer problem to implement in Java, where I want the producer thread to run for a specific amount of time e.g. 1 day, putting objects in a BlockingQueue -specifically tweets, streamed from Twitter Streaming API via Twitter4j- and the consumer thread to consume these objects from the queue and write them to file. I've used the PC logic from Read the 30Million user id's one by one from the big file, where producer is the FileTask and consumer is the CPUTask (check first answer; my approach uses the same iterations/try-catch blocks with it). Of course I adapted the implementations accordingly. My main function is:

public static void main(String[] args) {
    ....

    final int threadCount = 2;

    // BlockingQueue with a capacity of 200
    BlockingQueue<Tweet> tweets = new ArrayBlockingQueue<>(200);

    // create thread pool with given size
    ExecutorService service = Executors.newFixedThreadPool(threadCount);

    Future<?> f = service.submit(new GathererTask(tweets));
    try {
        f.get(1,TimeUnit.MINUTES); // Give specific time to the GathererTask
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
        f.cancel(true); // Stop the Gatherer
    }

    try {
        service.submit(new FileTask(tweets)).get(); // Wait til FileTask completes
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }

    service.shutdownNow();

   try {
        service.awaitTermination(7, TimeUnit.DAYS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

现在，问题在于，尽管它确实流化了这些tweet并将其写入文件，但它从未终止，也从未到达f.cancel(true)部分.要使其正常工作，我应该更改些什么?另外，您能否在回答中解释线程逻辑在这里出了什么问题，所以我可以从我的错误中学到什么?预先谢谢你.

Now, the problem is that, although it does stream the tweets and writes them to file, it never terminates and never gets to the f.cancel(true) part. What should I change for it to work properly? Also, could you explain in your answer what went wrong here with the thread logic, so I learn from my mistake? Thank you in advance.

这些是我的PC类的run()功能:

These are the run() functions of my PC classes:

制作人:

@Override
    public void run() {
        StatusListener listener = new StatusListener(){
            public void onStatus(Status status) {
                try {
                    tweets.put(new Tweet(status.getText(),status.getCreatedAt(),status.getUser().getName(),status.getHashtagEntities()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    Thread.currentTread.interrupt(); // Also tried this command
            }
        }

        public void onException(Exception ex) {
            ex.printStackTrace();
        }
    };
    twitterStream.addListener(listener);
    ... // More Twitter4j commands
}

消费者:

public void run() {
    Tweet tweet;
    try(PrintWriter out = new PrintWriter(new BufferedWriter(new FileWriter("out.csv", true)))) {
        while(true) {
            try {
                // block if the queue is empty
                tweet = tweets.take();
                writeTweetToFile(tweet,out);

            } catch (InterruptedException ex) {
                break; // GathererTask has completed
            }
        }
        // poll() returns null if the queue is empty
        while((tweet = tweets.poll()) != null) {
            writeTweetToFile(tweet,out);
        }

    } catch (IOException e) {
        e.printStackTrace();
    }
}

推荐答案

您应该检查Thread类是否正在处理InterruptedException，否则，它们将永远等待. 这可能会有所帮助.

You should check if your Thread classes are handling the InterruptedException, if not, they will wait forever. This might help.

这篇关于使用线程的Java生产者使用者永不终止的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！