多线程时获取 BlockingQueue 的大小

Question

我有一个多线程进程、5 个线程和另一个用作状态对象的线程，报告 BlockingQueue 的大小。问题是 Status 线程首先报告 100%，这是正确的，但随后直接变为 0%。

我想让它倒数百分比。

这是我的代码：

Thread[] workers = new Thread[NUMBER_OF_THREADS];
        for (int x = 0; x < NUMBER_OF_THREADS; x++) {
            workers[x] = new Thread(new S3ObjectDownloader(filesToDownload, currentYear));
            workers[x].start();
        }
        for (int x = 0; x < NUMBER_OF_THREADS; x++) {
            try {
                workers[x].join();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

下面是状态对象实例化：

int downloadSize = filesToDownload.size();
        Thread statusThread = new Thread(new Status(filesToDownload, currentYear,downloadSize,"DOWNLOADING..."));
        statusThread.start();

这是实际的 Status 对象运行方法：

public void run() {
    while(!queue.isEmpty()){
        try {
            float completion = (queue.size()*1)/this.queueSize;
            System.out.println(this.jobeName+" : "+this.conferenceYear+ " completion..."+MessageFormat.format("{0,number,#.##%}",completion));
            TimeUnit.SECONDS.sleep(30);;
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

我正在添加实际的 S3ObjectDownloader：

public void run() {
        //aws credentials
        this.s3 = new AmazonS3Client(new ClasspathPropertiesFileCredentialsProvider());

        //log4j configuration
        PropertyConfigurator.configure("/home/ubuntu/log4j.properties");

        //attempt to poll the queue
        while (!queue.isEmpty()) {

            String fileName = queue.poll() + ".mp4";
            String FULL_PATH = "best_of_ats/" + this.conferenceYear + "/videos/" + fileName;
            File f = new File("/home/ubuntu/" + fileName);

            if (fileName != null && !f.exists() && s3.doesObjectExist(BUCKET_NAME, FULL_PATH)) {
                OutputStream out = null;
                InputStream in = null;
                S3Object s3obj = null;


                    try {
                        s3obj = s3.getObject(this.BUCKET_NAME,
                                FULL_PATH);
                        in = s3obj.getObjectContent();
                        //System.out.println("Downloading File " + FULL_PATH + "....");
                    } catch (AmazonS3Exception s3e) {
                        // s3e.printStackTrace();
                        //System.out.println("Problem downloading file..." + FULL_PATH);
                        s3e.printStackTrace();
                        logger.info("Problem with file..." + FULL_PATH);
                        continue;
                    }

                    try {
                        out = new FileOutputStream(new File(fileName));
                        int read = 0;
                        byte[] bytes = new byte[1024];
                        while ((read = in.read(bytes)) != -1) {
                            out.write(bytes, 0, read);
                        }

                        out.flush();
                        out.close();
                        in.close();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                        //System.out.println("problem writing output..." + FULL_PATH);
                        logger.info("problem writing output..." +FULL_PATH);
                        continue;
                    }



            }

        } // end while...
    }

这是状态 Class:

public class Status implements Runnable {

    private String conferenceYear;
    private Queue<String>queue;
    private int queueSize;
    private String jobeName;

    public Status(Queue<String> queue, String conferenceYear, int queueSize, String jobName){
        this.conferenceYear = conferenceYear;
        this.queue = queue;
        this.queueSize = queueSize;
        this.jobeName = jobName;
    }

    @Override
    public void run() {
        while(!queue.isEmpty()){
            try {
                float completion = (queue.size()*1)/this.queueSize;
                System.out.println(this.jobeName+" : "+this.conferenceYear+ " completion..."+MessageFormat.format("{0,number,#.##%}",completion));
                TimeUnit.SECONDS.sleep(30);;
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    }

}

调用class:

public static void main(String[] args) {

        BlockingQueue<String> filesToDownload = new LinkedBlockingDeque<String>(1024);
        BlockingQueue<String> filesToPreview = new LinkedBlockingDeque<String>(1024);
        BlockingQueue<String> filesToUpload = new LinkedBlockingDeque<String>(1024);
        String currentYear = String.valueOf(Calendar.getInstance().get(Calendar.YEAR));

        // DB connection.
        ATSStoreDB db = new ATSStoreDB();
        PreparedStatement st = null;
        Connection conn = null;
        conn = db.getConnection();

        // get ids from ats_store.products.
        try {
            st = conn.prepareStatement(sql);
            st.setString(1, currentYear);
            ResultSet rs = st.executeQuery();
            // add each id to IDS.
            while (rs.next()) {
                filesToDownload.add(rs.getString("product_id"));
                filesToPreview.add(rs.getString("product_id"));
                filesToUpload.add(rs.getString("product_id"));
            }
            conn.close();

        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        /*
         * Distribute IDS to several threads.
         */

        //start up the Status Object class.
        int downloadSize = filesToDownload.size();
        Thread statusThread = new Thread(new Status(filesToDownload, currentYear,downloadSize,"DOWNLOADING..."));
        statusThread.start();

        /**
         * download the files
         */

        Thread[] workers = new Thread[NUMBER_OF_THREADS];
        for (int x = 0; x < NUMBER_OF_THREADS; x++) {
            workers[x] = new Thread(new S3ObjectDownloader(filesToDownload, currentYear));
            workers[x].start();
        }
        for (int x = 0; x < NUMBER_OF_THREADS; x++) {
            try {
                workers[x].join();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

        /**
         * create previews
         */
        int previewSize = filesToPreview.size();
        statusThread = new Thread(new Status(filesToPreview, currentYear,previewSize,"PREVIEWING..."));
        statusThread.start();


        workers = new Thread[NUMBER_OF_THREADS];
        for (int x = 0; x < NUMBER_OF_THREADS; x++) {
            workers[x] = new Thread(new Worker(filesToPreview, currentYear));
            workers[x].start();
        }
        for (int x = 0; x < NUMBER_OF_THREADS; x++) {
            try {
                workers[x].join();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }


    }

Answer 1

我可以立即发现您的代码中的一个问题：

float completion = (queue.size()*1)/this.queueSize;

*1有什么意义？ queue.size() 和 this.queueSize 都是整数。您正在将整数除法变成...整数除法。一个好的编译器可能会立即优化它。你可能打算写类似

的东西

float completion = (queue.size() * 1.0f) / this.queueSize;

多线程时获取 BlockingQueue 的大小

Get Size of BlockingQueue while Multithreading

java

multithreading

blockingqueue