国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

Java多線程編程實戰:模擬大量數據同步

elliott_hu / 2186人閱讀

摘要:所以得出結論需要分配較多的線程進行讀數據,較少的線程進行寫數據。注意多線程編程對實際環境和需求有很大的依賴,需要根據實際的需求情況對各個參數做調整。

背景

最近對于 Java 多線程做了一段時間的學習,筆者一直認為,學習東西就是要應用到實際的業務需求中的。否則要么無法深入理解,要么硬生生地套用技術只是達到炫技的效果。

不過筆者仍舊認為自己對于多線程掌握不夠熟練,不敢輕易應用到生產代碼中。這就按照平時工作中遇到的實際問題,腦補了一個很可能存在的業務場景:

已知某公司管理著 1000 個微信服務號,每個服務號有 1w ~ 50w 粉絲不等。假設該公司每天都需要將所有微信服務號的粉絲數據通過調用微信 API 的方式更新到本地數據庫。

需求分析

對此需求進行分析,主要存在以下問題:

單個服務號獲取粉絲 id,只能每次 1w 按順序拉取

微信的 API 對于服務商的并發請求數量有限制

單個服務號獲取粉絲 id,只能每次 1w 按順序拉取。這個問題決定了單個公眾號在拉取粉絲 id 上,無法分配給多個線程執行。

微信的 API 對于服務商的并發請求數量有限制。這點最容易被忽略,如果我們同時有過多的請求,則會導致接口被封禁。這里可以通過信號量來控制同時執行的線程數量。

為了盡快完成數據同步,根據實際情況:整個數據同步可分為讀數據和寫數據兩個部分。讀數據是通過 API 獲取,走網絡 IO,速度較慢;寫數據是寫到數據庫,速度較快。所以得出結論:需要分配較多的線程進行讀數據,較少的線程進行寫數據。

設計要點

首先,我們需要確定開啟多少個線程(在生產中往往是使用線程池),線程數量需要根據服務器性能來決定,這里我們定為 40 個讀取數據線程(將 1000 個公眾號分為 40 份,分別在 40 個線程中執行),1個寫入數據線程。(具體開多少個線程,取決于線程池的容量,以及可以分配給此業務的數量。具體的數字需要根據實際情況測試得出,比服務器閾值低一些較好。當然,配置允許范圍內越大越好)

其次,考慮到微信對于 API 并發請求的限制,需要限制同時執行的線程數,使用java.util.concurrent.Semaphore進行控制,這里我們限制為 20 個(具體的信號量憑證數,取決于同一時間能夠執行的線程,跟 API 限制,服務器性能有關)。

然后,我們需要知道數據何時讀取、寫入完畢,以控制程序邏輯以及終止程序,這里我們使用java.util.concurrent.CountDownLatch進行控制。

最后,我們需要一個數據結構,用來在多個線程中共享處理的數據,此處同步數據的場景非常適合使用隊列,這里我們使用線程安全的java.util.concurrent.ConcurrentLinkedQueue來進行處理。(需要注意的是,在實際開發中,隊列不能夠無限制地增長,這將會很快消耗掉內存,我們需要根據實際情況對隊列長度做控制。例如,可以通過控制讀取線程數和寫入線程數的比例來控制隊列的長度)

模擬代碼

由于本文重點關注多線程的使用,模擬代碼只體現多線程操作的方法。代碼里添加了大量的注釋,方便各位讀者閱讀理解。

JDK:1.8

import java.util.Arrays;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * N個線程向隊列添加數據
 * 一個線程消費隊列數據
 */
public class QueueTest {
    private static List data = Arrays.asList("a", "b", "c", "d", "e");

    private static final int OFFER_COUNT = 40; // 開啟的線程數量

    private static Semaphore semaphore = new Semaphore(20); // 同一時間執行的線程數量(大多用于控制API調用次數或數據庫查詢連接數)

    public static void main(String[] args) throws InterruptedException {
        Queue queue = new ConcurrentLinkedQueue<>(); // 處理隊列,需要處理的數據,放置到此隊列中

        CountDownLatch offerLatch = new CountDownLatch(OFFER_COUNT); // offer線程latch,每完成一個,latch減一,lacth的count為0時表示offer處理完畢
        CountDownLatch pollLatch = new CountDownLatch(1); // poll線程latch,latch的count為0時,表示poll處理完畢

        Runnable offerRunnable = () -> {
            try {
                semaphore.acquire(); // 信號量控制
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            try {
                for (String datum : data) {
                    queue.offer(datum);
                    TimeUnit.SECONDS.sleep(2); // 模擬取數據很慢的情況
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 在finally中執行latch.countDown()以及信號量釋放,避免因異常導致沒有正常釋放
                offerLatch.countDown();
                semaphore.release();
            }
        };

        Runnable pollRunnable = () -> {
            int count = 0;
            try {
                while (offerLatch.getCount() > 0 || queue.size() > 0) { // 只要offer的latch未執行完,或queue仍舊有數據,則繼續循環
                    String poll = queue.poll();
                    if (poll != null) {
                        System.out.println(poll);
                        count++;
                    }
                    // 無論是否poll到數據,均暫停一小段時間,可降低CPU消耗
                    TimeUnit.MILLISECONDS.sleep(100);
                }
                System.out.println("total count:" + count);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 在finally中執行latch.countDown(),避免因異常導致沒有正常釋放
                pollLatch.countDown();
            }
        };

        // 啟動線程(生產環境中建議使用線程池)
        new Thread(pollRunnable).start(); // 啟動一個poll線程
        for (int i = 0; i < OFFER_COUNT; i++) {
            new Thread(offerRunnable).start();
        } // 模擬取數據很慢,需要開啟40個線程處理

        // latch等待,會block主線程直到latch的count為0
        offerLatch.await();
        pollLatch.await();

        System.out.println("===the end===");
    }
}

到這里,本文結束。以上是筆者腦補的一個常見需求的解決方案。

注意:多線程編程對實際環境和需求有很大的依賴,需要根據實際的需求情況對各個參數做調整。實際在使用中,需要盡量模擬生產環境的數據情況來進行測試,對服務器執行期間的并發數,CPU、內存、網絡 IO、磁盤 IO 做好觀察。并適當地調低并發數,以給服務器留有處理其他請求的余量。

文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。

轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/73249.html

相關文章

  • 后端好書閱讀與推薦(續三)

    摘要:后端好書閱讀與推薦系列文章后端好書閱讀與推薦后端好書閱讀與推薦續后端好書閱讀與推薦續二后端好書閱讀與推薦續三這里依然記錄一下每本書的亮點與自己讀書心得和體會,分享并求拍磚。然后又請求封鎖,當釋放了上的封鎖之后,系統又批準了的請求一直等待。 后端好書閱讀與推薦系列文章:后端好書閱讀與推薦后端好書閱讀與推薦(續)后端好書閱讀與推薦(續二)后端好書閱讀與推薦(續三) 這里依然記錄一下每本書的...

    ckllj 評論0 收藏0
  • 后端好書閱讀與推薦(續三)

    摘要:后端好書閱讀與推薦系列文章后端好書閱讀與推薦后端好書閱讀與推薦續后端好書閱讀與推薦續二后端好書閱讀與推薦續三這里依然記錄一下每本書的亮點與自己讀書心得和體會,分享并求拍磚。然后又請求封鎖,當釋放了上的封鎖之后,系統又批準了的請求一直等待。 后端好書閱讀與推薦系列文章:后端好書閱讀與推薦后端好書閱讀與推薦(續)后端好書閱讀與推薦(續二)后端好書閱讀與推薦(續三) 這里依然記錄一下每本書的...

    jcc 評論0 收藏0
  • 后端好書閱讀與推薦(續三)

    摘要:后端好書閱讀與推薦系列文章后端好書閱讀與推薦后端好書閱讀與推薦續后端好書閱讀與推薦續二后端好書閱讀與推薦續三這里依然記錄一下每本書的亮點與自己讀書心得和體會,分享并求拍磚。然后又請求封鎖,當釋放了上的封鎖之后,系統又批準了的請求一直等待。 后端好書閱讀與推薦系列文章:后端好書閱讀與推薦后端好書閱讀與推薦(續)后端好書閱讀與推薦(續二)后端好書閱讀與推薦(續三) 這里依然記錄一下每本書的...

    lauren_liuling 評論0 收藏0
  • Java線程學習(七)并發編程中一些問題

    摘要:相比與其他操作系統包括其他類系統有很多的優點,其中有一項就是,其上下文切換和模式切換的時間消耗非常少。因為多線程競爭鎖時會引起上下文切換。減少線程的使用。很多編程語言中都有協程。所以如何避免死鎖的產生,在我們使用并發編程時至關重要。 系列文章傳送門: Java多線程學習(一)Java多線程入門 Java多線程學習(二)synchronized關鍵字(1) java多線程學習(二)syn...

    dingding199389 評論0 收藏0

發表評論

0條評論

最新活動
閱讀需要支付1元查看
<