Sejauh manakah pengaturcaraan serentak berbilang benang Java meningkatkan kecekapan pemprosesan data?

Dalam senario kerja, kami menghadapi keperluan sedemikian: untuk mengemas kini maklumat berkaitan model lain berdasarkan alamat IP hos. Keperluan adalah sangat mudah dan hanya melibatkan pertanyaan pautan pangkalan data umum dan operasi kemas kini Walau bagaimanapun, semasa proses pelaksanaan pengekodan, didapati bahawa disebabkan bilangan hos yang banyak, ia mengambil masa yang lama untuk membuat pertanyaan dan kemas kini kira-kira 30-40 saat untuk memanggil antara muka masa min untuk menyelesaikan operasi.

Oleh itu, untuk memendekkan masa pelaksanaan kaedah antara muka dengan berkesan, pertimbangkan untuk menggunakan kaedah pengaturcaraan serentak berbilang benang, mengambil kesempatan daripada keupayaan pelaksanaan selari pemproses berbilang teras, dan memproses data secara tak segerak, yang boleh sangat memendekkan masa pelaksanaan dan meningkatkan kecekapan Pelaksanaan.

Kolam benang boleh guna semula dengan bilangan benang tetap FixedThreadPool digunakan di sini dan alat kawalan proses serentak yang disediakan oleh kelas alat konkurensi CountDownLatch digunakan bersama untuk memastikan operasi normal semasa berbilang benang pengaturcaraan serentak:

• Pertama , dapatkan bilangan utas CPU mesin yang sedang berjalan melalui kaedah Runtime.getRuntime().availableProcessors(), yang digunakan untuk menetapkan bilangan utas seterusnya dalam kolam benang tetap.

• Kedua, , tentukan ciri-ciri tugasan tersebut Jika ia adalah tugasan yang intensif pengkomputeran, tetapkan bilangan utas kepada CPU 线程数+1 tugas intensif IO, tetapkan bilangan utas Untuk 2 * CPU 线程数, kerana interaksi kerap dengan pangkalan data diperlukan dalam kaedah, ia adalah tugas intensif IO.

• Selepas itu, kumpulan dan potong data Setiap utas memproses satu data terkumpul Bilangan kumpulan terkumpul adalah konsisten dengan bilangan utas dan pembilang juga dibuat. Objek CountDownLatch, panggil pembina, dan mulakan nilai parameter kepada bilangan utas untuk memastikan bahawa utas utama menunggu semua sub-utas selesai dijalankan sebelum melakukan operasi seterusnya.

• Kemudian , panggil kaedah executorService.execute(), tulis semula kaedah run untuk menulis logik perniagaan dan kod pemprosesan data, ingat untuk mengurangkan kaunter sebanyak 1 selepas melaksanakan operasi benang semasa. Akhir sekali, apabila semua benang kanak-kanak telah selesai dilaksanakan, tutup kumpulan benang.

Selepas meninggalkan kod logik perniagaan dalam senario kerja, contoh kaedah pemprosesan umum adalah seperti berikut:

public ResponseData updateHostDept() {
		// ...
		List<Map> hostMapList = mongoTemplate.find(query, Map.class, "host");
        // split the hostMapList for the following multi-threads task
        // return the number of logical CPUs
        int processorsNum = Runtime.getRuntime().availableProcessors();
        // set the threadNum as 2*(the number of logical CPUs) for handling IO Tasks,
        // if Computing Tasks set the threadNum as (the number of logical  CPUs) + 1
        int threadNum = processorsNum * 2;  
        // the number of each group data 
        int eachGroupNum = hostMapList.size() / threadNum; 
        List<List<Map>> groupList = new ArrayList<>();
        for (int i = 0; i < threadNum; i++) {
            int start = i * eachGroupNum;
            if (i == threadNum - 1) {
                int end = mapList.size();
                groupList.add(hostMapList.subList(start, end));
            } else {
                int end = (i+1) * eachGroupNum;
                groupList.add(hostMapList.subList(start, end));
            }
        }
        // update data by using multi-threads asynchronously
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum/2);
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        for (List<Map> group : groupList) {
            executorService.execute(()->{
                try {
                    for (Map map : group) {
                    	// update the data in mongodb
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                	// let counter minus one 
                    countDownLatch.countDown();  
                }
            });
        }
        try {
        	// main thread donnot execute until all child threads finish
            countDownLatch.await();  
        } catch (Exception e) {
            e.printStackTrace();
        }
        // remember to shutdown the threadPool
        executorService.shutdown();  
        return ResponseData.success();
}

Kemudian selepas menggunakan strategi kemas kini tak segerak berbilang benang, Anggaran masa yang diperlukan untuk memanggil antara muka telah menurun daripada 30-40 min kepada 8-10 min, meningkatkan kecekapan pelaksanaan.

Perlu diambil perhatian bahawa newFixedThreadPool yang digunakan di sini untuk mencipta kumpulan benang mempunyai kecacatan bahawa baris gilir menyekatnya ialah baris gilir tidak terhad secara lalai, dan nilai lalai ialah Integer.MAX_VALUE yang sangat berkemungkinan Menyebabkan masalah OOM. Oleh itu, anda secara amnya boleh menggunakan ThreadPoolExecutor untuk mencipta kumpulan benang, dan anda boleh menentukan bilangan utas dalam baris gilir menunggu untuk mengelakkan masalah OOM.

public ResponseData updateHostDept() {
		// ...
		List<Map> hostMapList = mongoTemplate.find(query, Map.class, "host");
        // split the hostMapList for the following multi-threads task
        // return the number of logical CPUs
        int processorsNum = Runtime.getRuntime().availableProcessors();
        // set the threadNum as 2*(the number of logical CPUs) for handling IO Tasks,
        // if Computing Tasks set the threadNum as (the number of logical  CPUs) + 1
        int threadNum = processorsNum * 2;  
        // the number of each group data 
        int eachGroupNum = hostMapList.size() / threadNum; 
        List<List<Map>> groupList = new ArrayList<>();
        for (int i = 0; i < threadNum; i++) {
            int start = i * eachGroupNum;
            if (i == threadNum - 1) {
                int end = mapList.size();
                groupList.add(hostMapList.subList(start, end));
            } else {
                int end = (i+1) * eachGroupNum;
                groupList.add(hostMapList.subList(start, end));
            }
        }
        // update data by using multi-threads asynchronously
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 8, 30L, TimeUnit.SECONDS, 
                new ArrayBlockingQueue<>(100));
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        for (List<Map> group : groupList) {
            executor.execute(()->{
                try {
                    for (Map map : group) {
                    	// update the data in mongodb
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                	// let counter minus one 
                    countDownLatch.countDown();  
                }
            });
        }
        try {
        	// main thread donnot execute until all child threads finish
            countDownLatch.await();  
        } catch (Exception e) {
            e.printStackTrace();
        }
        // remember to shutdown the threadPool
        executor.shutdown();  
        return ResponseData.success();
}

Dalam kod di atas, bilangan utas teras dan bilangan maksimum utas masing-masing adalah 5 dan 8 Mereka tidak ditetapkan kepada nilai yang sangat besar, kerana jika ia ditetapkan kepada nilai yang besar , gangguan yang kerap antara benang akan berlaku juga akan meningkatkan penggunaan masa, tetapi tidak akan memaksimumkan kelebihan berbilang benang. Bagi cara memilih parameter yang sesuai, ia perlu ditentukan berdasarkan parameter mesin dan jenis tugas.

Akhir sekali, jika anda ingin mendapatkan bilangan utas CPU mesin melalui kaedah bukan pengekodan, ia juga sangat mudah Dalam sistem Windows, anda boleh melihat bilangan utas CPU melalui Pengurus Tugas dan pilih "Prestasi". , seperti yang ditunjukkan dalam gambar di bawah:

Seperti yang anda lihat daripada gambar di atas, teras dalam mesin saya ialah lapan CPU, tetapi satu fizikal Teras CPU boleh disimulasikan melalui teknologi hyper-threading ke dalam dua utas CPU logik, jadi mesin saya menyokong 8 teras dan 16 utas.

Atas ialah kandungan terperinci Sejauh manakah pengaturcaraan serentak berbilang benang Java meningkatkan kecekapan pemprosesan data?. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!