標籤: 台北網頁設計公司

繼法國之後，英國政府宣布2040 年禁售汽柴油車

access_timePosted on by admin

汽柴油車真的要走入歷史了嗎？繼法國宣布2040 年禁售汽柴油車後，英國政府為解決空氣污染問題，準備從30 億英鎊抗空汙的資金中提撥2.55 億英鎊協助委員會加快地方措施，以應對柴油車輛的污染，終極目標也將在2040 年前禁售汽柴油車。

英國獨立報(The Independent) 報導，空氣污染與英國每年約4 萬人過早死亡有關，運輸也佔溫室氣體排放量的很大一部分。英國政府先前推出的抗空汙計畫版本被環保人士反對，認為力道太弱，無法達到歐盟的排放標準，英國最高法院要求7 月31 日前英國政府必須制定新的計畫，以降低有害二氧化氮的排放量，而就在法院規定的截止日前，英國政府宣布2040 年汽柴油車禁令。

英國追隨法國腳步頒布禁售令，顯示向電動車轉型的速度正在加快，BMW 宣布計畫推出Mini 電動車版，將在英國牛津進行組裝，Volvo 也宣布清潔能源車計畫。

英國政府還將討論柴油車報廢計劃的執行細節。英國環保倡議者認為，這項計劃應包括政府資助且強制性的清潔空氣區，對進入高空氣污染地區的高污染車輛收取費用。對清潔空氣區設立收費制度被視為是最有效打擊二氧化氮污染的政策，而柴油車是排量二氧化氮的禍首。

但是英國政府對此有疑慮，認為這是懲罰柴油車駕駛，畢竟原本認為柴油車的碳排量比汽車少，因此鼓勵消費者購買。英國政府傾向改裝巴士等交通工具，降低排放量，或改變道路佈局，甚至改變速度和重新編排交通號誌等功能，使交通流量更加順暢，減少污染。

英國政府發言人表示，不該責怪柴油車駕駛，為了幫助他們改用清潔車，政府會討論針對性的報廢計劃，支持受本地計劃影響的駕駛。但英國在野黨不滿意這種溫和的作風，呼籲柴油車禁售令應該提前至2025 年，並提出廢止計劃幫駕駛轉換成更環保的車輛。

印度2030 年實現全電動車目標

除歐洲國家之外，受嚴重空汙困擾的印度動作也非常積極，印度政府計畫2030 年實現全電動車目標，而印度政府此舉並不只是為了抗空汙，還可減少燃料進口費用。印度重工業部和印度國家研究院正在製定促進電動汽車發展政策，主要是朝降低成本提高價格誘因做起，現在印度對混合動力車與電動車提供補貼，初期也會補貼業者度過轉型期，為2030 年禁售汽柴油車鋪路。

未來3 年印度將大規模佈建充電基礎設施與電池交換計畫，目前只有印度只有電動車廠  Mahindra Electric 在印度提供全電動車，最近Anand Mahindra 執行長在社群媒體上邀請Tesla 到印度設置商店，Tesla 將在2017 年底之前進入印度市場，開放印度消費者訂購Model 3，明年可能開始展店。

除了電動汽車，混合動力汽車市場處於更好的市場位置，豐田、Volvo 和BMW 等幾家製造商在印度提供混合動力車或插電式混合動力車。特斯拉和日產也宣布進在印度市場堆出Model 3和Leaf。若印度政府的計劃順利進行，印度將成為全球電動車品牌的一級戰場。

（合作媒體：。圖片出處：public domain CC0）

本站聲明:網站內容來源於EnergyTrend https://www.energytrend.com.tw/ev/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

動手造輪子：實現簡單的 EventQueue

access_timePosted on by admin

動手造輪子：實現簡單的 EventQueue

Intro

最近項目里有遇到一些併發的問題，想實現一個隊列來將併發的請求一個一個串行處理，可以理解為使用消息隊列處理併發問題，之前實現過一個簡單的 EventBus，於是想在 EventBus 的基礎上改造一下，加一個隊列，改造成類似消息隊列的處理模式。消息的處理（Consumer）直接使用 .netcore 里的 IHostedService 來實現了一個簡單的後台任務處理。

初步設計

  • Event 抽象的事件
  • EventHandler 處理 Event 的方法
  • EventStore 保存訂閱 Event 的 EventHandler
  • EventQueue 保存 Event 的隊列
  • EventPublisher 發布 Event
  • EventConsumer 處理 Event 隊列里的 Event
  • EventSubscriptionManager 管理訂閱 Event 的 EventHandler

實現代碼

EventBase 定義了基本事件信息，事件發生時間以及事件的id:

public abstract class EventBase
{
    [JsonProperty]
    public DateTimeOffset EventAt { get; private set; }

    [JsonProperty]
    public string EventId { get; private set; }

    protected EventBase()
    {
      this.EventId = GuidIdGenerator.Instance.NewId();
      this.EventAt = DateTimeOffset.UtcNow;
    }

    [JsonConstructor]
    public EventBase(string eventId, DateTimeOffset eventAt)
    {
      this.EventId = eventId;
      this.EventAt = eventAt;
    }
}

EventHandler 定義：

public interface IEventHandler
{
    Task Handle(IEventBase @event);
}

public interface IEventHandler<in TEvent> : IEventHandler where TEvent : IEventBase
{
    Task Handle(TEvent @event);
}

public class EventHandlerBase<TEvent> : IEventHandler<TEvent> where TEvent : EventBase
{
    public virtual Task Handle(TEvent @event)
    {
        return Task.CompletedTask;
    }

    public Task Handle(IEventBase @event)
    {
        return Handle(@event as TEvent);
    }
}

EventStore：

public class EventStore
{
    private readonly Dictionary<Type, Type> _eventHandlers = new Dictionary<Type, Type>();

    public void Add<TEvent, TEventHandler>() where TEventHandler : IEventHandler<TEvent> where TEvent : EventBase
    {
        _eventHandlers.Add(typeof(TEvent), typeof(TEventHandler));
    }

    public object GetEventHandler(Type eventType, IServiceProvider serviceProvider)
    {
        if (eventType == null || !_eventHandlers.TryGetValue(eventType, out var handlerType) || handlerType == null)
        {
            return null;
        }
        return serviceProvider.GetService(handlerType);
    }

    public object GetEventHandler(EventBase eventBase, IServiceProvider serviceProvider) =>
        GetEventHandler(eventBase.GetType(), serviceProvider);

    public object GetEventHandler<TEvent>(IServiceProvider serviceProvider) where TEvent : EventBase =>
        GetEventHandler(typeof(TEvent), serviceProvider);
}

EventQueue 定義：

public class EventQueue
{
    private readonly ConcurrentDictionary<string, ConcurrentQueue<EventBase>> _eventQueues =
        new ConcurrentDictionary<string, ConcurrentQueue<EventBase>>();

    public ICollection<string> Queues => _eventQueues.Keys;

    public void Enqueue<TEvent>(string queueName, TEvent @event) where TEvent : EventBase
    {
        var queue = _eventQueues.GetOrAdd(queueName, q => new ConcurrentQueue<EventBase>());
        queue.Enqueue(@event);
    }

    public bool TryDequeue(string queueName, out EventBase @event)
    {
        var queue = _eventQueues.GetOrAdd(queueName, q => new ConcurrentQueue<EventBase>());
        return queue.TryDequeue(out @event);
    }

    public bool TryRemoveQueue(string queueName)
    {
        return _eventQueues.TryRemove(queueName, out _);
    }

    public bool ContainsQueue(string queueName) => _eventQueues.ContainsKey(queueName);

    public ConcurrentQueue<EventBase> this[string queueName] => _eventQueues[queueName];
}

EventPublisher:

public interface IEventPublisher
{
    Task Publish<TEvent>(string queueName, TEvent @event)
        where TEvent : EventBase;
}
public class EventPublisher : IEventPublisher
{
    private readonly EventQueue _eventQueue;

    public EventPublisher(EventQueue eventQueue)
    {
        _eventQueue = eventQueue;
    }

    public Task Publish<TEvent>(string queueName, TEvent @event)
        where TEvent : EventBase
    {
        _eventQueue.Enqueue(queueName, @event);
        return Task.CompletedTask;
    }
}

EventSubscriptionManager:

public interface IEventSubscriptionManager
{
    void Subscribe<TEvent, TEventHandler>()
        where TEvent : EventBase
        where TEventHandler : IEventHandler<TEvent>;
}

public class EventSubscriptionManager : IEventSubscriptionManager
{
    private readonly EventStore _eventStore;

    public EventSubscriptionManager(EventStore eventStore)
    {
        _eventStore = eventStore;
    }

    public void Subscribe<TEvent, TEventHandler>()
        where TEvent : EventBase
        where TEventHandler : IEventHandler<TEvent>
    {
        _eventStore.Add<TEvent, TEventHandler>();
    }
}

EventConsumer:

public class EventConsumer : BackgroundService
{
    private readonly EventQueue _eventQueue;
    private readonly EventStore _eventStore;
    private readonly int maxSemaphoreCount = 256;
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger _logger;

    public EventConsumer(EventQueue eventQueue, EventStore eventStore, IConfiguration configuration, ILogger<EventConsumer> logger, IServiceProvider serviceProvider)
    {
        _eventQueue = eventQueue;
        _eventStore = eventStore;
        _logger = logger;
        _serviceProvider = serviceProvider;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using (var semaphore = new SemaphoreSlim(Environment.ProcessorCount, maxSemaphoreCount))
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                var queues = _eventQueue.Queues;
                if (queues.Count > 0)
                {
                    await Task.WhenAll(
                    queues
                        .Select(async queueName =>
                        {
                            if (!_eventQueue.ContainsQueue(queueName))
                            {
                                return;
                            }
                            try
                            {
                                await semaphore.WaitAsync(stoppingToken);
                                //
                                if (_eventQueue.TryDequeue(queueName, out var @event))
                                {
                                    var eventHandler = _eventStore.GetEventHandler(@event, _serviceProvider);
                                    if (eventHandler is IEventHandler handler)
                                    {
                                        _logger.LogInformation(
                                            "handler {handlerType} begin to handle event {eventType}, eventId: {eventId}, eventInfo: {eventInfo}",
                                            eventHandler.GetType().FullName, @event.GetType().FullName,
                                            @event.EventId, JsonConvert.SerializeObject(@event));

                                        try
                                        {
                                            await handler.Handle(@event);
                                        }
                                        catch (Exception e)
                                        {
                                            _logger.LogError(e, "event  {eventId}  handled exception", @event.EventId);
                                        }
                                        finally
                                        {
                                            _logger.LogInformation("event {eventId} handled", @event.EventId);
                                        }
                                    }
                                    else
                                    {
                                        _logger.LogWarning(
                                            "no event handler registered for event {eventType}, eventId: {eventId}, eventInfo: {eventInfo}",
                                            @event.GetType().FullName, @event.EventId,
                                            JsonConvert.SerializeObject(@event));
                                    }
                                }
                            }
                            catch (Exception ex)
                            {
                                _logger.LogError(ex, "error running EventConsumer");
                            }
                            finally
                            {
                                semaphore.Release();
                            }
                        })
                );
                }

                await Task.Delay(50, stoppingToken);
            }
        }
    }
}

為了方便使用定義了一個 Event 擴展方法：

public static IServiceCollection AddEvent(this IServiceCollection services)
{
    services.TryAddSingleton<EventStore>();
    services.TryAddSingleton<EventQueue>();
    services.TryAddSingleton<IEventPublisher, EventPublisher>();
    services.TryAddSingleton<IEventSubscriptionManager, EventSubscriptionManager>();

    services.AddSingleton<IHostedService, EventConsumer>();
    return services;
}

使用示例

定義 PageViewEvent 記錄請求信息：

public class PageViewEvent : EventBase
{
    public string Path { get; set; }
}

這裏作為示例只記錄了請求的Path信息，實際使用可以增加更多需要記錄的信息

定義 PageViewEventHandler，處理 PageViewEvent

public class PageViewEventHandler : EventHandlerBase<PageViewEvent>
{
    private readonly ILogger _logger;

    public PageViewEventHandler(ILogger<PageViewEventHandler> logger)
    {
        _logger = logger;
    }

    public override Task Handle(PageViewEvent @event)
    {
        _logger.LogInformation($"handle pageViewEvent: {JsonConvert.SerializeObject(@event)}");
        return Task.CompletedTask;
    }
}

這個 handler 里什麼都沒做只是輸出一個日誌

這個示例項目定義了一個記錄請求路徑的事件以及一個發布請求記錄事件的中間件

// 發布 Event 的中間件
app.Use(async (context, next) =>
{
    var eventPublisher = context.RequestServices.GetRequiredService<IEventPublisher>();
    await eventPublisher.Publish("pageView", new PageViewEvent() { Path = context.Request.Path.Value });
    await next();
});

Startup 配置：

public void ConfigureServices(IServiceCollection services)
{
    // ...
    services.AddEvent();
    services.AddSingleton<PageViewEventHandler>();// 註冊 Handler
}

// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
public void Configure(IApplicationBuilder app, IHostingEnvironment env, IEventSubscriptionManager eventSubscriptionManager)
{
    eventSubscriptionManager.Subscribe<PageViewEvent, PageViewEventHandler>();
    app.Use(async (context, next) =>
    {
        var eventPublisher = context.RequestServices.GetRequiredService<IEventPublisher>();
        await eventPublisher.Publish("pageView", new PageViewEvent() { Path = context.Request.Path.Value });
        await next();
    });
    // ...
}

使用效果:

More

注：只是一個初步設計，基本可以實現功能，還是有些不足，實際應用的話還有一些要考慮的事情

  1. Consumer 消息邏輯，現在的實現有些問題，我們的應用場景目前比較簡單還可以滿足，如果事件比較多就會而且每個事件可能處理需要的時間長短不一樣，會導致在一個批次中執行的 Event 中已經完成的事件要等待其他還沒完成的事件完成之後才能繼續取下一個事件，理想的消費模式應該是各個隊列相互獨立，在同一個隊列中保持順序消費即可
  2. 上面示例的 EventStore 的實現只是簡單的實現了一個事件一個 Handler 的處理情況，實際業務場景中很可能會有一個事件需要多個 Handler 的情況
  3. 這個實現是基於內存的，如果要在分佈式場景下使用就不適用了，需要自己實現一下基於redis或者數據庫的以滿足分佈式的需求
  4. and more…

上面所有的代碼可以在 Github 上獲取，示例項目 Github 地址：

Reference

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

Freemarker + xml 實現Java導出word

access_timePosted on by admin

前言

最近做了一個調查問卷導出的功能，需求是將維護的題目，答案，導出成word，參考了幾種方案之後，選擇功能強大的freemarker+固定格式之後的wordxml實現導出功能。導出word的代碼是可以直接復用的，於是在此貼出，並進行總結，方便大家拿走。

實現過程概覽

先在word上，調整好自己想要的樣子。然後存為xml文件。保存為freemarker模板，以ftl後綴結尾。將需要替換的變量使用freemarker的語法進行替換。最終將數據準備好，和模板進行渲染，生成文件並返回給瀏覽器流。

詳細的實現過程

準備好word的樣式

我們新建一個word，我們應該使用Microsoft office，如果使用wps可能會造成樣式有些不兼容。在新建的office中，設置好我們的表格樣式。我們的調查問卷涉及到四種類型，單選，多選，填空，簡答。我們做出四種類型的示例。

樣式沒有問題后，我們選擇另存為word xml 2003版本。將會生成一個xml文件。

格式化xml，並用freemarker語法替換xml

我們可以先下載一個工具 firstobject xml editor，這個可以幫助我們查看xml，同時方便我們定位我們需要改的位置。
複製過去之後，按f8可以將其進行格式化，左側是標籤，右側是內容，我們只需要關注w:body即可。

像右側的調查問卷這個就是個標題，我們實際渲染的時候應該將其進行替換，比如我們的程序數據map中，有title屬性，我們想要這裏展示，我們就使用語法${title}即可。

freemarker的具體語法，可以參考freemarker的問題，在這裏我給出幾個簡單的例子。
比如我們將所有的數據放置在dataList中，所以我們需要判斷，dataList是不是空，是空，我們不應該進行下面的邏輯，不是空，我們應該先循環題目是必須的，答案是需要根據類型進行再次循環的。語法參考文檔，這裏不再贅述。

程序端引入freemarker

<dependency>
    <groupId>org.freemarker</groupId>
    <artifactId>freemarker</artifactId>
</dependency>

將我們的flt文件放在resources下的templates下。

後端代碼實現

此代碼可以復用，在此貼出

public class WordUtils {

    private static Configuration configuration = null;
    private static final String templateFolder = WordUtils.class.getClassLoader().getResource("").getPath()+"/templates/word";
    static {
        configuration = new Configuration();
        configuration.setDefaultEncoding("utf-8");
        try {
            configuration.setDirectoryForTemplateLoading(new File(templateFolder));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     *  @Description:導出word，傳入request，response，map就是值，title是導出問卷名，ftl是你要使用的模板名
     */
    public static void exportWord(HttpServletRequest request, HttpServletResponse response, Map map, String title, String ftlFile) throws Exception {
        Template freemarkerTemplate = configuration.getTemplate(ftlFile);
        File file = null;
        InputStream fin = null;
        ServletOutputStream out = null;
        try {
            file = createDocFile(map,freemarkerTemplate);
            fin = new FileInputStream(file);
            String fileName = title + ".doc";
            response.setCharacterEncoding("utf-8");
            response.setContentType("application/msword");
            response.setHeader("Content-Disposition", "attachment;filename="
             +fileName);
            out = response.getOutputStream();
            byte[] buffer = new byte[512];  
            int bytesToRead = -1;
            while((bytesToRead = fin.read(buffer)) != -1) {
                out.write(buffer, 0, bytesToRead);
            }
        }finally {
            if(fin != null) fin.close();
            if(out != null) out.close();
            if(file != null) file.delete(); 
        }
    }

    /**
     *  @Description:創建doc文件
     */
    private static File createDocFile(Map<?, ?> dataMap, Template template) {
        File file = new File("init.doc");
        try {
            Writer writer = new OutputStreamWriter(new FileOutputStream(file), "utf-8");
            template.process(dataMap, writer);
            writer.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return file;
    }

}

有了工具類后，我們準備好我們的map數據。map裏面的數據大家可以自行定義。然後調用utils中的導出方法即可。

WordUtils.exportWord(request, response, dataMap, "word", "demo.ftl");

結語

至此已經結束了，十分的好用，有疑問的話，可以評論交流。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

.NET進階篇06-async異步、thread多線程3

access_timePosted on by admin

知識需要不斷積累、總結和沉澱，思考和寫作是成長的催化劑

梯子

一、任務Task

System.Threading.Tasks在.NET4引入，前麵線程的API太多了，控制不方便，而ThreadPool控制能力又太弱，比如做線程的延續、阻塞、取消、超時等功能不太方便，所以Task就抽象了線程功能，在後台使用ThreadPool

1、啟動任務

可以使用TaskFactory類或Task類的構造函數和Start()方法，委託可以提供帶有一個Object類型的輸入參數，所以可以給任務傳遞任意數據，還漏了一個常用的Task.Run

TaskFactory taskFactory = new TaskFactory();
taskFactory.StartNew(() => 
{
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
Task.Factory.StartNew(() =>
{
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
Task task = new Task(() =>
{
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
task.Start();

只有Task類實例方式需要Start()去啟動任務，當然可以RunSynchronously()來同步執行任務，主線程會等待，就是用主線程來執行這個task任務

Task task = new Task(() =>
{
    Thread.Sleep(10000);
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
task.RunSynchronously();

2、阻塞延續

在Thread中我們使用join來阻塞等待，在多個Thread時進行控制就不太方便。Task中我們使用實例方法Wait阻塞單個任務或靜態方法WaitAll和WaitAny阻塞多個任務

var task = new Task(() =>
{
    Thread.Sleep(5*1000);
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
var task2 = new Task(() =>
{
    Thread.Sleep(10 * 1000);
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
task.Start();
task2.Start();
//task.Wait();//單任務等待
//Task.WaitAny(task, task2);//任何一個任務完成就繼續
Task.WaitAll(task, task2);//任務都完成才繼續

如果不希望阻塞主線程，實現當一個任務或幾個任務完成后執行別的任務，可以使用Task靜態方法WhenAll和WhenAny,他們將返回一個Task，但這個Task不允許你控制，將會在滿足WhenAll和WhenAny里任務完成時自動完成，然後調用Task的ContinueWith方法，就可以在一個任務完成后緊跟開始另一個任務

Task.WhenAll(task, task2).ContinueWith((t) =>
{
    Console.WriteLine($"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});

Task.Factory工廠中也存在類似ContinueWhenAll和ContinueWhenAny

3、任務層次結構

不僅可以在一個任務結束后執行另一個任務，也可以在一個任務內啟動一個任務，這就啟動了一個父子層次結構

var parentTask = new Task(()=> 
{
    Console.WriteLine($"parentId={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
    Thread.Sleep(5*1000);
    var childTask = new Task(() =>
    {
        Thread.Sleep(10 * 1000);
        Console.WriteLine($"childId={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}")
    });
    childTask.Start();
});
parentTask.Start();

如果父任務在子任務之前結束，父任務的狀態為WaitingForChildrenToComplete，當子任務也完成時，父任務的狀態就變為RanToCompletion，當然，在創建任務時指定TaskCreationOptions枚舉參數，可以控制任務的創建和執行的可選行為

4、枚舉參數

簡單介紹下創建任務中的TaskCreationOptions枚舉參數，創建任務時我們可以提供TaskCreationOptions枚舉參數，用於控制任務的創建和執行的可選行為的標誌

  1. AttachedToParent：指定將任務附加到任務層次結構中的某個父級，意思就是建立父子關係，父任務必須等待子任務完成才可以繼續執行。和WaitAll效果一樣。上面例子如果在創建子任務時指定TaskCreationOptions.AttachedToParent，那麼父任務wait時也會等子任務的結束
  2. DenyChildAttach：不讓子任務附加到父任務上
  3. LongRunning：指定是長時間運行任務，如果事先知道這個任務會耗時比較長，建議設置此項。這樣，Task調度器會創建Thread線程，而不使用ThreadPool線程。因為你長時間佔用ThreadPool線程不還，那它可能必要時會在線程池中開啟新的線程，造成調度壓力
  4. PreferFairness：盡可能公平的安排任務，這意味着較早安排的任務將更可能較早運行，而較晚安排運行的任務將更可能較晚運行。實際通過把任務放到線程池的全局隊列中，讓工作線程去爭搶，默認是在本地隊列中。

另一個枚舉參數是ContinueWith方法中的TaskContinuationOptions枚舉參數，它除了擁有幾個和上面同樣功能的枚舉值外，還擁有控制任務的取消延續等功能

  1. LazyCancellation：在延續取消的情況下，防止延續的完成直到完成先前的任務。什麼意思呢？
CancellationTokenSource source = new CancellationTokenSource();
source.Cancel();
var task1 = new Task(() => 
{
    Console.WriteLine($"task1 id={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
var task2 = task1.ContinueWith(t =>
{
    Console.WriteLine($"task2 id={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
},source.Token);
var task3 = task2.ContinueWith(t =>
{
    Console.WriteLine($"task3 id={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
});
task1.Start();

上面例子我們企圖task1->task2->task3順序執行，然後通過CancellationToken來取消task2的執行。結果會是怎樣呢？結果task1和task3會并行執行（task3也是會執行的，而且是和task1并行，等於原來的一條鏈變成了兩條鏈），然後我們嘗試使用LazyCancellation，

var task2 = task1.ContinueWith(t =>
{
    Console.WriteLine($"task2 id={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
},source.Token,TaskContinuationOptions.LazyCancellation,TaskScheduler.Current);

這樣，將會在task1執行完成后，task2才去判斷source.Token，為Cancel就不執行，接下來執行task3就保證了原來的順序

  1. ExecuteSynchronously：指定應同步執行延續任務，比如上例中，在延續任務task2中指定此參數，則task2會使用執行task1的線程來執行，這樣防止線程切換，可以做些共有資源的訪問。不指定的話就隨機，但也能也用到task1的線程
  2. NotOnRanToCompletion：延續任務必須在前面任務非完成狀態下執行
  3. OnlyOnRanToCompletion：延續任務必須在前面任務完成狀態才能執行
  4. NotOnFaulted，OnlyOnCanceled，OnlyOnFaulted等等

5、任務取消

在上篇使用Thread時，我們使用一個變量isStop標記是否取消任務，這種訪問共享變量的方式難免會出問題。task中提出CancellationTokenSource類專門處理任務取消，常見用法看下面代碼註釋

CancellationTokenSource source = new CancellationTokenSource();//構造函數中也可指定延遲取消
//註冊一個取消時調用的委託
source.Token.Register(() =>
{
    Console.WriteLine("當前source已經取消，可以在這裏做一些其他事情（比如資源清理）...");
});
var task1 = new Task(() => 
{
    while (!source.IsCancellationRequested)
    {
        Console.WriteLine($"task1 id={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}");
    }
},source.Token);
task1.Start();
//source.Cancel();//取消
source.CancelAfter(1000);//延時取消

6、任務結果

讓子線程返回結果，可以將信息寫入到線程安全的共享變量中去，或則使用可以返回結果的任務。使用Task的泛型版本Task<TResult>，就可以定義返回結果的任務。Task是繼承自Task的，Result獲取結果時是要阻塞等待直到任務完成返回結果的，內部判斷沒有完成則wait。通過TaskStatus屬性可獲得此任務的狀態是啟動、運行、異常還是取消等

var task = new Task<string>(() =>
{
     return "hello ketty";
});
task.Start();
string result = task.Result;

7、異常

可以使用AggregateException來接受任務中的異常信息，這是一個聚合異常繼承自Exception，可以遍歷獲取包含的所有異常，以及進行異常處理，決定是否繼續往上拋異常等

var task = Task.Factory.StartNew(() =>
{
    var childTask1 = Task.Factory.StartNew(() =>
    {
        throw new Exception("childTask1異常...");
    },TaskCreationOptions.AttachedToParent);
    var childTask12= Task.Factory.StartNew(() =>
    {
        throw new Exception("childTask2異常...");
    }, TaskCreationOptions.AttachedToParent);
});
try
{
    try
    {
        task.Wait();
    }
    catch (AggregateException ex)
    {
        foreach (var item in ex.InnerExceptions)
        {
            Console.WriteLine($"message{item.InnerException.Message}");
        }
        ex.Handle(x =>
        {
            if (x.InnerException.Message == "childTask1異常...")
            {
                return true;//異常被處理，不繼續往上拋了
            }
            return false;
        });
    }
}
catch (Exception ex)
{
    throw;
}

二、并行Parallel

1、Parallel.For()、Parallel.ForEach()

在.NET4中，另一個新增的抽象的線程時Parallel類。這個類定義了并行的for和foreach的靜態方法。Parallel.For()和Parallel.ForEach()方法多次調用一個方法，而Parallel.Invoke()方法允許同時調用不同的方法。首先Parallel是會阻塞主線程的，它將讓主線程也參与到任務中
Parallel.For()類似於for允許語句，并行迭代同一個方法，迭代順序沒有保證的

ParallelLoopResult result = Parallel.For(0, 10, i =>
{
    Console.WriteLine($"{i} task:{Task.CurrentId} thread:{Thread.CurrentThread.ManagedThreadId}");
});
Console.WriteLine(result.IsCompleted);

也可以提前中斷Parallel.For()方法。For()方法的一個重載版本接受Action<int,parallelloopstate style=”font-size: inherit; color: inherit; line-height: inherit; margin: 0px; padding: 0px;”>類型參數。一般不使用，像下面這樣，本想大於5就停止，但實際也可能有大於5的任務已經在跑了。可以通過ParallelOptions傳入允許最大線程數以及取消Token等

ParallelLoopResult result = Parallel.For(0, 10, new ParallelOptions() { MaxDegreeOfParallelism = 8 },(i,loop) =>
{
    Console.WriteLine($"{i} task:{Task.CurrentId} thread:{Thread.CurrentThread.ManagedThreadId}");
    if (i > 5)
    {
        loop.Break();
    }
});

2、Parallel.For<TLocal>

For還有一個高級泛型版本，相當於并行的聚合計算

ParallelLoopResult For<TLocal>(int fromInclusive, int toExclusive, Func<TLocal> localInit, Func<int, ParallelLoopState, TLocal, TLocal> body, Action<TLocal> localFinally);

像下面這樣我們求0…100的和，第三個參數更定一個種子初始值，第四個參數迭代累計，最後聚合

int totalNum = 0;
Parallel.For<int>(0, 100, () => { return 0; }, (current, loop, total) =>
{
    total += current;
    return total;
}, (total) =>
{
    Interlocked.Add(ref totalNum, total);
});

上面For用來處理數組數據，ForEach()方法用來處理非數組的數據任務，比如字典數據繼承自IEnumerable的集合等

3、Parallel.Invoke()

Parallel.Invoke()則可以并行調用不同的方法，參數傳遞一個Action的委託數組

Parallel.Invoke(() => { Console.WriteLine($"方法1 thread:{Thread.CurrentThread.ManagedThreadId}"); }
    , () => { Console.WriteLine($"方法2 thread:{Thread.CurrentThread.ManagedThreadId}"); }
    , () => { Console.WriteLine($"方法3 thread:{Thread.CurrentThread.ManagedThreadId}"); });

4、PLinq

Plinq，為了能夠達到最大的靈活度，linq有了并行版本。使用也很簡單，只需要將原始集合AsParallel就轉換為支持并行化的查詢。也可以AsOrdered來順序執行，取消Token，強制并行等

var nums = Enumerable.Range(0, 100);
var query = from n in nums.AsParallel()
            select new
            {
                thread=$"tid={Thread.CurrentThread.ManagedThreadId},datetime={DateTime.Now}"
            };

三、異步等待AsyncAwait

異步編程模型，可能還需要大篇幅來學習，這裏先介紹下基本用法，內在本質需要用ILSpy反編譯來看，以後可能要分專題總結。文末先給幾個參考資料，有興趣自己闊以先琢磨琢磨鴨

1、簡單使用

這是.NET4.5開始提供的一對語法糖，使得可以較簡便的使用異步編程。async用在方法定義前面，await只能寫在帶有async標記的方法中，任何方法都可以增加async，一般成對出現，只有async沒有意義，只有await會報錯，請先看下面的示例

private static async void AsyncTest()
{
    //主線程執行
    Console.WriteLine($"before await ThreadId={Thread.CurrentThread.ManagedThreadId}");
    TaskFactory taskFactory = new TaskFactory();
    Task task = taskFactory.StartNew(() =>
    {
        Thread.Sleep(3000);
        Console.WriteLine($"task ThreadId={Thread.CurrentThread.ManagedThreadId}");
    });
    await task;//主線程到這裏就返回了，執行主線程任務
    //子線程執行，其實是封裝成委託，在task之後成為回調（編譯器功能  狀態機實現） 後面相當於task.ContinueWith()
    //這個回調的線程是不確定的：可能是主線程  可能是子線程  也可能是其他線程，在winform中是主線程
    Console.WriteLine($"after await ThreadId={Thread.CurrentThread.ManagedThreadId}");
}

一般使用async都會讓方法返回一個Task的，像下面這樣複雜一點的

private static async Task<string> AsyncTest2()
{
    Console.WriteLine($"before await ThreadId={Thread.CurrentThread.ManagedThreadId}");
    TaskFactory taskFactory = new TaskFactory();
    string x = await taskFactory.StartNew(() =>
      {
          Thread.Sleep(3000);
          Console.WriteLine($"task ThreadId={Thread.CurrentThread.ManagedThreadId}");
          return "task over";
      });

    Console.WriteLine($"after await ThreadId={Thread.CurrentThread.ManagedThreadId}");
    return x;
}

通過var reslult = AsyncTest2().Result;調用即可。但注意如果調用Wait或Result的代碼位於UI線程，Task的實際執行在其他線程，其需要返回UI線程則會造成死鎖，所以應該Async all the way

2、優雅

從上面簡單示例中可以看出異步編程的執行邏輯：主線程A邏輯->異步任務線程B邏輯->主線程C邏輯。
異步方法的返回類型只能是void、Task、Task。示例中異步方法的返回值類型是Task，通常void也不推薦使用，沒有返回值直接用Task就是

上一篇也大概了解到如果我們要在任務中更新UI，需要調用Invoke通知UI線程來更新，代碼看起來像下面這樣，在一個任務後去更新UI

private void button1_Click(object sender, EventArgs e)
{
    var ResultTask = Task.Run(() => {
        Thread.Sleep(5000);
        return "任務完成";
    });
    ResultTask.ContinueWith((r)=> 
    {
        textBox1.Invoke(() => {
            textBox1.Text = r.Result;
        });
    });
}

如果使用async/await會看起來像這樣，是不是優雅了許多。以看似同步編程的方式實現異步

private async void button1_Click(object sender, EventArgs e)
{
    var t = Task.Run(() => {
        Thread.Sleep(5000);
        return "任務完成";
    });
    textBox1.Text = await t;
}

3、最後

在.NET 4.5中引入的Async和Await兩個新的關鍵字后，用戶能以一種簡潔直觀的方式實現異步編程。甚至都不需要改變代碼的邏輯結構，就能將原來的同步函數改造為異步函數。
在內部實現上，Async和Await這兩個關鍵字由編譯器轉換為狀態機，通過System.Threading.Tasks中的并行類實現代碼的異步執行。

字數有點多了，我的能力也就高考作文800字能寫的出奇好。看了很多異步編程，腦袋有點炸，等消化后再輸出一次，技藝不足，只能用輸出倒逼輸入了，下一篇會是線程安全集合、鎖問題、同步問題，基於事件的異步模式等

Search the fucking web
Read the fucking maunal

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

利用Python學習線性代數 — 1.1 線性方程組

access_timePosted on by admin

利用Python學習線性代數 — 1.1 線性方程組

系列，

本節實現的主要功能函數，在源碼文件中，後續章節將作為基本功能調用。

線性方程

線性方程組由一個或多個線性方程組成，如
\[ \begin{array}\\ x_1 – 2 x_2 &= -1\\ -x_1 + 3 x_2 &= 3 \end{array} \]

求包含兩個變量兩個線性方程的方程組的解，等價於求兩條直線的交點。
這裏可以畫出書圖1-1和1-2的線性方程組的圖形。
通過改變線性方程的參數，觀察圖形，體會兩個方程對應直線平行、相交、重合三種可能。

那麼，怎麼畫二元線性方程的直線呢？

方法是這樣的：
假如方程是 \(a x_1 + b x_2 = c\) 的形式，可以寫成 \(x_2 = (c – a x_1) / b\)。
在以 \(x_1\) 和\(x_2\)為兩個軸的直角坐標系中，\(x_1\)取一組值，如 \((-3, -2.9, -2.8, \dots, 2.9, 3.0)\)，
計算相應的 \(x_2\)，然後把所有點 \((x_1, x_2)\) 連起來成為一條線。
當 \(b\) 為 \(0\) 時， 則在\(x_1 = c / a\)處畫一條垂直線。

# 引入Numpy和 Matplotlib庫
import numpy as np
from matplotlib import pyplot as plt

Matplotlib 是Python中使用較多的可視化庫，這裏只用到了它的一些基本功能。

def draw_line(a, b, c, start=-4, 
              stop=5, step=0.01):
    """根據線性方程參數繪製一條直線"""
    # 如果b為0，則畫一條垂線
    if np.isclose(b, 0):
        plt.vlines(start, stop, c / a)
    else: # 否則畫 y = (c - a*x) / b
        xs = np.arange(start, stop, step)
        plt.plot(xs, (c - a*xs)/b)
# 1.1 圖1-1
draw_line(1, -2, -1)
draw_line(-1, 3, 3)

def draw_lines(augmented, start=-4, 
              stop=5, step=0.01):
    """給定增廣矩陣，畫兩條線."""
    plt.figure()
    for equation in augmented:
        draw_line(*equation, start, stop, step)
    plt.show()
# Fig. 1-1
# 增廣矩陣用二維數組表示 
# [[1, -2, -1], [-1, 3, 3]]
# 這些数字對應圖1-1對應方程的各項係數
draw_lines([[1, -2, -1], [-1, 3, 3]])

# Fig. 1-2
draw_lines([[1, -2, -2], [-1, 2, 3]])
# Fig. 1-3
draw_lines([[1, -2, -1], [-1, 2, 1]])

  • 建議：改變這些係數，觀察直線，體會兩條直線相交、平行和重合的情況

例如

draw_lines([[1, -2, -2], [-1, 2, 9]])

如果對Numpy比較熟悉，則可以採用更簡潔的方式實現上述繪圖功能。
在計算多條直線方程時，可以利用向量編程的方式，用更少的代碼實現。

def draw_lines(augmented, start=-4, 
               stop=5, step=0.01):
    """Draw lines represented by augmented matrix on 2-d plane."""
    am = np.asarray(augmented)
    xs = np.arange(start, stop, step).reshape([1, -1])
    # 同時計算兩條直線的y值
    ys = (am[:, [-1]] - am[:, [1]]*xs) / am[:, [0]]
    for y in ys:
        plt.plot(xs[0], y)
    plt.show()

矩陣記號

矩陣是一個數表，在程序中通常用二維數組表示，例如

# 嵌套列表表示矩陣
matrix = [[1, -2, 1, 0],
          [0, 2, -8, 8],
          [5, 0, -5, 10]]
matrix
[[1, -2, 1, 0], [0, 2, -8, 8], [5, 0, -5, 10]]

實際工程和研究實踐中，往往會採用一些專門的數值計算庫，簡化和加速計算。
Numpy庫是Python中數值計算的常用庫。
在Numpy中，多維數組類型稱為ndarray，可以理解為n dimensional array。
例如

# Numpy ndarray 表示矩陣
matrix = np.array([[1, -2, 1, 0],
                    [0, 2, -8, 8],
                    [5, 0, -5, 10]])
matrix
array([[ 1, -2,  1,  0],
       [ 0,  2, -8,  8],
       [ 5,  0, -5, 10]])

解線性方程組

本節解線性方程組的方法是 高斯消元法，利用了三種基本行變換。

  1. 把某個方程換成它與另一個方程的倍數的和；
  2. 交換兩個方程的位置；
  3. 某個方程的所有項乘以一個非零項。

假設線性方程的增廣矩陣是\(A\)，其第\(i\)行\(j\)列的元素是\(a_{ij}\)。
消元法的基本步驟是：

  • 增廣矩陣中有 \(n\) 行，該方法的每一步處理一行。
    1. 在第\(i\)步，該方法處理第\(i\)行
      • 若\(a_{ii}\)為0，則在剩餘行 \(\{j| j \in (i, n]\}\)中選擇絕對值最大的行\(a_{ij}\)
        • 若\(a_{ij}\)為0，返回第1步。
        • 否則利用變換2，交換\(A\)的第\(i\)和\(j\)行。
    2. 利用行變換3，第\(i\)行所有元素除以\(a_{ii}\)，使第 \(i\) 個方程的第 \(i\)個 係數為1
    3. 利用行變換1，\(i\)之後的行減去第\(i\)行的倍數，使這些行的第 \(i\) 列為0

為了理解這些步驟的實現，這裏先按書中的例1一步步計算和展示，然後再總結成完整的函數。
例1的增廣矩陣是

\[ \left[ \begin{array} &1 & -2 & 1 & 0\\ 0 & 2 & -8 & 8\\ 5 & 0 & -5 & 10 \end{array} \right] \]

# 增廣矩陣
A = np.array([[1, -2, 1, 0],
              [0, 2, -8, 8],
              [5, 0, -5, 10]])
# 行號從0開始，處理第0行
i = 0
# 利用變換3，將第i行的 a_ii 轉成1。這裏a_00已經是1，所不用動
# 然後利用變換1，把第1行第0列，第2行第0列都減成0。
# 這裏僅需考慮i列之後的元素，因為i列之前的元素已經是0
#   即第1行減去第0行的0倍
#   而第2行減去第0行的5倍
A[i+1:, i:] = A[i+1:, i:] - A[i+1:, [i]] * A[i, i:]
A
array([[  1,  -2,   1,   0],
       [  0,   2,  -8,   8],
       [  0,  10, -10,  10]])
i = 1
# 利用變換3，將第i行的 a_ii 轉成1。
A[i] = A[i] / A[i, i]
A
array([[  1,  -2,   1,   0],
       [  0,   1,  -4,   4],
       [  0,  10, -10,  10]])
# 然後利用變換1，把第2行第i列減成0。
A[i+1:, i:] = A[i+1:, i:] - A[i+1:, [i]] * A[i, i:]
A
array([[  1,  -2,   1,   0],
       [  0,   1,  -4,   4],
       [  0,   0,  30, -30]])
i = 2
# 利用變換3，將第i行的 a_ii 轉成1。
A[i] = A[i] / A[i, i]
A
array([[ 1, -2,  1,  0],
       [ 0,  1, -4,  4],
       [ 0,  0,  1, -1]])

消元法的前向過程就結束了，我們可以總結成一個函數

def eliminate_forward(augmented): 
    """
    消元法的前向過程.
    
    返回行階梯形，以及先導元素的坐標（主元位置）
    """
    A = np.asarray(augmented, dtype=np.float64)
    # row number of the last row
    pivots = []
    i, j = 0, 0
    while i < A.shape[0] and j < A.shape[1]:
        A[i] = A[i] / A[i, j]
        if (i + 1) < A.shape[0]: # 除最後一行外
            A[i+1:, j:] = A[i+1:, j:] - A[i+1:, [j]] * A[i, j:]
        pivots.append((i, j))
        i += 1
        j += 1
    return A, pivots

這裡有兩個細節值得注意

  1. 先導元素 \(a_{ij}\)，不一定是在主對角線位置，即 \(i\) 不一定等於\(j\).
  2. 最後一行只需要用變換3把先導元素轉為1，沒有剩餘行需要轉換
# 測試一個增廣矩陣，例1
A = np.array([[1, -2, 1, 0],
              [0, 2, -8, 8],
              [5, 0, -5, 10]])
A, pivots = eliminate_forward(A)
print(A)
print(pivots)
[[ 1. -2.  1.  0.]
 [ 0.  1. -4.  4.]
 [ 0.  0.  1. -1.]]
[(0, 0), (1, 1), (2, 2)]

消元法的後向過程則更簡單一些，對於每一個主元(這裏就是前面的\(a_{ii}\))，將其所在的列都用變換1，使其它行對應的列為0.

for i, j in reversed(pivots):
    A[:i, j:] = A[:i, j:] - A[[i], j:] * A[:i, [j]] 
A
array([[ 1.,  0.,  0.,  1.],
       [ 0.,  1.,  0.,  0.],
       [ 0.,  0.,  1., -1.]])
def eliminate_backward(simplified, pivots):
    """消元法的後向過程."""
    A = np.asarray(simplified)
    for i, j in reversed(pivots):
        A[:i, j:] = A[:i, j:] - A[[i], j:] * A[:i, [j]] 
    return A

至此，結合 eliminate_forward 和eliminate_backward函數，可以解形如例1的線性方程。

然而，存在如例3的線性方程，在eliminate_forward算法進行的某一步，主元為0，需要利用變換2交換兩行。
交換行時，可以選擇剩餘行中，選擇當前主元列不為0的任意行，與當前行交換。
這裏每次都採用剩餘行中，當前主元列絕對值最大的行。
補上行交換的前向過程函數如下

def eliminate_forward(augmented): 
    """消元法的前向過程"""
    A = np.asarray(augmented, dtype=np.float64)
    # row number of the last row
    pivots = []
    i, j = 0, 0
    while i < A.shape[0] and j < A.shape[1]:
        # if pivot is zero, exchange rows
        if np.isclose(A[i, j], 0):
            if (i + 1) < A.shape[0]:
                max_k = i + 1 + np.argmax(np.abs(A[i+1:, i]))
            if (i + 1) >= A.shape[0] or np.isclose(A[max_k, i], 0):
                j += 1
                continue
            A[[i, max_k]] = A[[max_k, i]]
        A[i] = A[i] / A[i, j]
        if (i + 1) < A.shape[0]:
            A[i+1:, j:] = A[i+1:, j:] - A[i+1:, [j]] * A[i, j:]
        pivots.append((i, j))
        i += 1
        j += 1
    return A, pivots

行交換時，有一種特殊情況，即剩餘所有行的主元列都沒有非零元素。
這種情況下，在當前列的右側尋找不為零的列，作為新的主元列。

# 用例3測試eliminate_forward
aug = [[0, 1, -4, 8],
       [2, -3, 2, 1],
       [4, -8, 12, 1]]
echelon, pivots = eliminate_forward(aug)
print(echelon)
print(pivots)
[[ 1.   -2.    3.    0.25]
 [ 0.    1.   -4.    0.5 ]
 [ 0.    0.    0.    1.  ]]
[(0, 0), (1, 1), (2, 3)]

例3化簡的結果與書上略有不同，由行交換策略不同引起，也說明同一個矩陣可能由多個階梯形。

結合上述的前向和後向過程，即可以給出一個完整的消元法實現

def eliminate(augmented):
    """
    利用消元法前向和後向步驟，化簡線性方程組.
    
    如果是矛盾方程組，則僅輸出前向化簡結果，並打印提示
    否則輸出簡化后的方程組，並輸出最後一列
    """
    print(np.asarray(augmented))
    A, pivots = eliminate_forward(augmented)
    print(" The echelon form is\n", A)
    print(" The pivots are: ", pivots)
    pivot_cols = {p[1] for p in pivots}
    simplified = eliminate_backward(A, pivots)
    if (A.shape[1]-1) in pivot_cols:
        print(" There is controdictory.\n", simplified)
    elif len(pivots) == (A.shape[1] -1):
        print(" Solution: ", simplified[:, -1])
        is_correct = solution_check(np.asarray(augmented), 
                            simplified[:, -1])
        print(" Is the solution correct? ", is_correct)
    else:
        print(" There are free variables.\n", simplified)
    print("-"*30)
eliminate(aug)
[[ 0  1 -4  8]
 [ 2 -3  2  1]
 [ 4 -8 12  1]]
 The echelon form is
 [[ 1.   -2.    3.    0.25]
 [ 0.    1.   -4.    0.5 ]
 [ 0.    0.    0.    1.  ]]
 The pivots are:  [(0, 0), (1, 1), (2, 3)]
 There is controdictory.
 [[ 1.  0. -5.  0.]
 [ 0.  1. -4.  0.]
 [ 0.  0.  0.  1.]]
------------------------------

利用 Sympy 驗證消元法實現的正確性

Python的符號計算庫Sympy，有化簡矩陣為行最簡型的方法，可以用來檢驗本節實現的代碼是否正確。

# 導入 sympy的 Matrix模塊
from sympy import Matrix
Matrix(aug).rref(simplify=True)
# 返回的是行最簡型和主元列的位置
(Matrix([
 [1, 0, -5, 0],
 [0, 1, -4, 0],
 [0, 0,  0, 1]]), (0, 1, 3))
echelon, pivots = eliminate_forward(aug)
simplified = eliminate_backward(echelon, pivots)
print(simplified, pivots)
# 輸出與上述rref一致
[[ 1.  0. -5.  0.]
 [ 0.  1. -4.  0.]
 [ 0.  0.  0.  1.]] [(0, 0), (1, 1), (2, 3)]

綜合前向和後向步驟，並結果的正確性

綜合前向和後向消元，就可以得到完整的消元法過程。
消元結束，如果沒有矛盾（最後一列不是主元列），基本變量數與未知數個數一致，則有唯一解，可以驗證解是否正確。
驗證的方法是將解與係數矩陣相乘，檢查與原方程的b列一致。

def solution_check(augmented, solution):
    # 係數矩陣與解相乘
    b = augmented[:, :-1] @ solution.reshape([-1, 1])
    b = b.reshape([-1])
    # 檢查乘積向量與b列一致
    return all(np.isclose(b - augmented[:, -1], np.zeros(len(b))))
def eliminate(augmented):
    from sympy import Matrix
    print(np.asarray(augmented))
    A, pivots = eliminate_forward(augmented)
    print(" The echelon form is\n", A)
    print(" The pivots are: ", pivots)
    pivot_cols = {p[1] for p in pivots}
    simplified = eliminate_backward(A, pivots)
    if (A.shape[1]-1) in pivot_cols: # 最後一列是主元列
        print(" There is controdictory.\n", simplified)
    elif len(pivots) == (A.shape[1] -1): # 唯一解
        is_correct = solution_check(np.asarray(augmented), 
                            simplified[:, -1])
        print(" Is the solution correct? ", is_correct)
        print(" Solution: \n", simplified)
    else: # 有自由變量
        print(" There are free variables.\n", simplified)
    print("-"*30)
    print("對比Sympy的rref結果")
    print(Matrix(augmented).rref(simplify=True))
    print("-"*30)

測試書中的例子

aug_1_1_1 = [[1, -2, 1, 0], 
             [0, 2, -8, 8], 
             [5, 0, -5, 10]]
eliminate(aug_1_1_1)
# 1.1 example 3
aug_1_1_3 = [[0, 1, -4, 8],
             [2, -3, 2, 1],
             [4, -8, 12, 1]]
eliminate(aug_1_1_3)
eliminate([[1, -6, 4, 0, -1],
           [0, 2, -7, 0, 4],
           [0, 0, 1, 2, -3],
           [0, 0, 3, 1, 6]])
eliminate([[0, -3, -6, 4, 9],
           [-1, -2, -1, 3, 1],
           [-2, -3, 0, 3, -1],
           [1, 4, 5, -9, -7]])

eliminate([[0, 3, -6, 6, 4, -5],
           [3, -7, 8, -5, 8, 9],
           [3, -9, 12, -9, 6, 15]])
[[ 1 -2  1  0]
 [ 0  2 -8  8]
 [ 5  0 -5 10]]
 The echelon form is
 [[ 1. -2.  1.  0.]
 [ 0.  1. -4.  4.]
 [ 0.  0.  1. -1.]]
 The pivots are:  [(0, 0), (1, 1), (2, 2)]
 Is the solution correct?  True
 Solution: 
 [[ 1.  0.  0.  1.]
 [ 0.  1.  0.  0.]
 [ 0.  0.  1. -1.]]
------------------------------
對比Sympy的rref結果
(Matrix([
[1, 0, 0,  1],
[0, 1, 0,  0],
[0, 0, 1, -1]]), (0, 1, 2))
------------------------------
[[ 0  1 -4  8]
 [ 2 -3  2  1]
 [ 4 -8 12  1]]
 The echelon form is
 [[ 1.   -2.    3.    0.25]
 [ 0.    1.   -4.    0.5 ]
 [ 0.    0.    0.    1.  ]]
 The pivots are:  [(0, 0), (1, 1), (2, 3)]
 There is controdictory.
 [[ 1.  0. -5.  0.]
 [ 0.  1. -4.  0.]
 [ 0.  0.  0.  1.]]
------------------------------
對比Sympy的rref結果
(Matrix([
[1, 0, -5, 0],
[0, 1, -4, 0],
[0, 0,  0, 1]]), (0, 1, 3))
------------------------------
[[ 1 -6  4  0 -1]
 [ 0  2 -7  0  4]
 [ 0  0  1  2 -3]
 [ 0  0  3  1  6]]
 The echelon form is
 [[ 1.  -6.   4.   0.  -1. ]
 [ 0.   1.  -3.5  0.   2. ]
 [ 0.   0.   1.   2.  -3. ]
 [-0.  -0.  -0.   1.  -3. ]]
 The pivots are:  [(0, 0), (1, 1), (2, 2), (3, 3)]
 Is the solution correct?  True
 Solution: 
 [[ 1.   0.   0.   0.  62. ]
 [ 0.   1.   0.   0.  12.5]
 [ 0.   0.   1.   0.   3. ]
 [-0.  -0.  -0.   1.  -3. ]]
------------------------------
對比Sympy的rref結果
(Matrix([
[1, 0, 0, 0,   62],
[0, 1, 0, 0, 25/2],
[0, 0, 1, 0,    3],
[0, 0, 0, 1,   -3]]), (0, 1, 2, 3))
------------------------------
[[ 0 -3 -6  4  9]
 [-1 -2 -1  3  1]
 [-2 -3  0  3 -1]
 [ 1  4  5 -9 -7]]
 The echelon form is
 [[ 1.   1.5 -0.  -1.5  0.5]
 [-0.   1.   2.  -3.  -3. ]
 [-0.  -0.  -0.   1.  -0. ]
 [ 0.   0.   0.   0.   0. ]]
 The pivots are:  [(0, 0), (1, 1), (2, 3)]
 There are free variables.
 [[ 1.  0. -3.  0.  5.]
 [-0.  1.  2.  0. -3.]
 [-0. -0. -0.  1. -0.]
 [ 0.  0.  0.  0.  0.]]
------------------------------
對比Sympy的rref結果
(Matrix([
[1, 0, -3, 0,  5],
[0, 1,  2, 0, -3],
[0, 0,  0, 1,  0],
[0, 0,  0, 0,  0]]), (0, 1, 3))
------------------------------
[[ 0  3 -6  6  4 -5]
 [ 3 -7  8 -5  8  9]
 [ 3 -9 12 -9  6 15]]
 The echelon form is
 [[ 1.         -2.33333333  2.66666667 -1.66666667  2.66666667  3.        ]
 [ 0.          1.         -2.          2.          1.33333333 -1.66666667]
 [ 0.          0.          0.          0.          1.          4.        ]]
 The pivots are:  [(0, 0), (1, 1), (2, 4)]
 There are free variables.
 [[  1.   0.  -2.   3.   0. -24.]
 [  0.   1.  -2.   2.   0.  -7.]
 [  0.   0.   0.   0.   1.   4.]]
------------------------------
對比Sympy的rref結果
(Matrix([
[1, 0, -2, 3, 0, -24],
[0, 1, -2, 2, 0,  -7],
[0, 0,  0, 0, 1,   4]]), (0, 1, 4))
------------------------------

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

Kafka冪等性原理及實現剖析

access_timePosted on by admin

1.概述

最近和一些同學交流的時候反饋說，在面試Kafka時，被問到Kafka組件組成部分、API使用、Consumer和Producer原理及作用等問題都能詳細作答。但是，問到一個平時不注意的問題，就是Kafka的冪等性，被卡主了。那麼，今天筆者就為大家來剖析一下Kafka的冪等性原理及實現。

2.內容

2.1 Kafka為啥需要冪等性？

Producer在生產發送消息時，難免會重複發送消息。Producer進行retry時會產生重試機制，發生消息重複發送。而引入冪等性后，重複發送只會生成一條有效的消息。Kafka作為分佈式消息系統，它的使用場景常見與分佈式系統中，比如消息推送系統、業務平台系統（如物流平台、銀行結算平台等）。以銀行結算平台來說，業務方作為上游把數據上報到銀行結算平台，如果一份數據被計算、處理多次，那麼產生的影響會很嚴重。

2.2 影響Kafka冪等性的因素有哪些？

在使用Kafka時，需要確保Exactly-Once語義。分佈式系統中，一些不可控因素有很多，比如網絡、OOM、FullGC等。在Kafka Broker確認Ack時，出現網絡異常、FullGC、OOM等問題時導致Ack超時，Producer會進行重複發送。可能出現的情況如下：

 

 

2.3 Kafka的冪等性是如何實現的？

Kafka為了實現冪等性，它在底層設計架構中引入了ProducerID和SequenceNumber。那這兩個概念的用途是什麼呢？

  • ProducerID：在每個新的Producer初始化時，會被分配一個唯一的ProducerID，這個ProducerID對客戶端使用者是不可見的。
  • SequenceNumber：對於每個ProducerID，Producer發送數據的每個Topic和Partition都對應一個從0開始單調遞增的SequenceNumber值。

2.3.1 冪等性引入之前的問題？

Kafka在引入冪等性之前，Producer向Broker發送消息，然後Broker將消息追加到消息流中后給Producer返回Ack信號值。實現流程如下：

 

上圖的實現流程是一種理想狀態下的消息發送情況，但是實際情況中，會出現各種不確定的因素，比如在Producer在發送給Broker的時候出現網絡異常。比如以下這種異常情況的出現：

 

上圖這種情況，當Producer第一次發送消息給Broker時，Broker將消息(x2,y2)追加到了消息流中，但是在返回Ack信號給Producer時失敗了（比如網絡異常） 。此時，Producer端觸發重試機制，將消息(x2,y2)重新發送給Broker，Broker接收到消息后，再次將該消息追加到消息流中，然後成功返回Ack信號給Producer。這樣下來，消息流中就被重複追加了兩條相同的(x2,y2)的消息。

2.3.2 冪等性引入之後解決了什麼問題？

面對這樣的問題，Kafka引入了冪等性。那麼冪等性是如何解決這類重複發送消息的問題的呢？下面我們可以先來看看流程圖：

 

 同樣，這是一種理想狀態下的發送流程。實際情況下，會有很多不確定的因素，比如Broker在發送Ack信號給Producer時出現網絡異常，導致發送失敗。異常情況如下圖所示：

 

 當Producer發送消息(x2,y2)給Broker時，Broker接收到消息並將其追加到消息流中。此時，Broker返回Ack信號給Producer時，發生異常導致Producer接收Ack信號失敗。對於Producer來說，會觸發重試機制，將消息(x2,y2)再次發送，但是，由於引入了冪等性，在每條消息中附帶了PID（ProducerID）和SequenceNumber。相同的PID和SequenceNumber發送給Broker，而之前Broker緩存過之前發送的相同的消息，那麼在消息流中的消息就只有一條(x2,y2)，不會出現重複發送的情況。

2.3.3 ProducerID是如何生成的？

客戶端在生成Producer時，會實例化如下代碼：

// 實例化一個Producer對象
Producer<String, String> producer = new KafkaProducer<>(props);

在org.apache.kafka.clients.producer.internals.Sender類中，在run()中有一個maybeWaitForPid()方法，用來生成一個ProducerID，實現代碼如下：

 private void maybeWaitForPid() {
        if (transactionState == null)
            return;

        while (!transactionState.hasPid()) {
            try {
                Node node = awaitLeastLoadedNodeReady(requestTimeout);
                if (node != null) {
                    ClientResponse response = sendAndAwaitInitPidRequest(node);
                    if (response.hasResponse() && (response.responseBody() instanceof InitPidResponse)) {
                        InitPidResponse initPidResponse = (InitPidResponse) response.responseBody();
                        transactionState.setPidAndEpoch(initPidResponse.producerId(), initPidResponse.epoch());
                    } else {
                        log.error("Received an unexpected response type for an InitPidRequest from {}. " +
                                "We will back off and try again.", node);
                    }
                } else {
                    log.debug("Could not find an available broker to send InitPidRequest to. " +
                            "We will back off and try again.");
                }
            } catch (Exception e) {
                log.warn("Received an exception while trying to get a pid. Will back off and retry.", e);
            }
            log.trace("Retry InitPidRequest in {}ms.", retryBackoffMs);
            time.sleep(retryBackoffMs);
            metadata.requestUpdate();
        }
    }

3.事務

與冪等性有關的另外一個特性就是事務。Kafka中的事務與數據庫的事務類似，Kafka中的事務屬性是指一系列的Producer生產消息和消費消息提交Offsets的操作在一個事務中，即原子性操作。對應的結果是同時成功或者同時失敗。

這裏需要與數據庫中事務進行區別，操作數據庫中的事務指一系列的增刪查改，對Kafka來說，操作事務是指一系列的生產和消費等原子性操作。

3.1 Kafka引入事務的用途？

在事務屬性引入之前，先引入Producer的冪等性，它的作用為：

  • Producer多次發送消息可以封裝成一個原子性操作，即同時成功，或者同時失敗；
  • 消費者&生產者模式下，因為Consumer在Commit Offsets出現問題時，導致重複消費消息時，Producer重複生產消息。需要將這個模式下Consumer的Commit Offsets操作和Producer一系列生產消息的操作封裝成一個原子性操作。

產生的場景有：

比如，在Consumer中Commit Offsets時，當Consumer在消費完成時Commit的Offsets為100（假設最近一次Commit的Offsets為50），那麼執行觸發Balance時，其他Consumer就會重複消費消息（消費的Offsets介於50~100之間的消息）。

3.2 事務提供了哪些可使用的API？

Producer提供了五種事務方法，它們分別是：initTransactions()、beginTransaction()、sendOffsetsToTransaction()、commitTransaction()、abortTransaction()，代碼定義在org.apache.kafka.clients.producer.Producer<K,V>接口中，具體定義接口如下：

// 初始化事務，需要注意確保transation.id屬性被分配
void initTransactions();

// 開啟事務
void beginTransaction() throws ProducerFencedException;

// 為Consumer提供的在事務內Commit Offsets的操作
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                              String consumerGroupId) throws ProducerFencedException;

// 提交事務
void commitTransaction() throws ProducerFencedException;

// 放棄事務，類似於回滾事務的操作
void abortTransaction() throws ProducerFencedException;

3.3 事務的實際應用場景有哪些？

在Kafka事務中，一個原子性操作，根據操作類型可以分為3種情況。情況如下：

  • 只有Producer生產消息，這種場景需要事務的介入；
  • 消費消息和生產消息並存，比如Consumer&Producer模式，這種場景是一般Kafka項目中比較常見的模式，需要事務介入；
  • 只有Consumer消費消息，這種操作在實際項目中意義不大，和手動Commit Offsets的結果一樣，而且這種場景不是事務的引入目的。

4.總結

Kafka的冪等性和事務是比較重要的特性，特別是在數據丟失和數據重複的問題上非常重要。Kafka引入冪等性，設計的原理也比較好理解。而事務與數據庫的事務特性類似，有數據庫使用的經驗對理解Kafka的事務也比較容易接受。

5.結束語

這篇博客就和大家分享到這裏，如果大家在研究學習的過程當中有什麼問題，可以加群進行討論或發送郵件給我，我會盡我所能為您解答，與君共勉！

另外，博主出書了《》和《》，喜歡的朋友或同學， 可以在公告欄那裡點擊購買鏈接購買博主的書進行學習，在此感謝大家的支持。關注下面公眾號，根據提示，可免費獲取書籍的教學視頻。 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

【原創】（十一）Linux內存管理slub分配器

access_timePosted on by admin

背景

  • Read the fucking source code! –By 魯迅
  • A picture is worth a thousand words. –By 高爾基

說明：

  1. Kernel版本：4.14
  2. ARM64處理器，Contex-A53，雙核
  3. 使用工具：Source Insight 3.5， Visio

1. 概述

之前的文章分析的都是基於頁面的內存分配，而小塊內存的分配和管理是通過塊分配器來實現的。目前內核中，有三種方式來實現小塊內存分配：slab, slub, slob，最先有slab分配器，slub/slob分配器是改進版，slob分配器適用於小內存嵌入式設備，而slub分配器目前已逐漸成為主流塊分配器。接下來的文章，就是以slub分配器為目標，進一步深入。

先來一個初印象：

2. 數據結構

有四個關鍵的數據結構：

  • struct kmem_cache：用於管理SLAB緩存，包括該緩存中對象的信息描述，per-CPU/Node管理slab頁面等；
    關鍵字段如下：
/*
 * Slab cache management.
 */
struct kmem_cache {
    struct kmem_cache_cpu __percpu *cpu_slab;       //每個CPU slab頁面
    /* Used for retriving partial slabs etc */
    unsigned long flags;
    unsigned long min_partial;
    int size;       /* The size of an object including meta data */
    int object_size;    /* The size of an object without meta data */
    int offset;     /* Free pointer offset. */
#ifdef CONFIG_SLUB_CPU_PARTIAL
    /* Number of per cpu partial objects to keep around */
    unsigned int cpu_partial;
#endif
    struct kmem_cache_order_objects oo;     //該結構體會描述申請頁面的order值，以及object的個數

    /* Allocation and freeing of slabs */
    struct kmem_cache_order_objects max;
    struct kmem_cache_order_objects min;
    gfp_t allocflags;   /* gfp flags to use on each alloc */
    int refcount;       /* Refcount for slab cache destroy */
    void (*ctor)(void *);           // 對象構造函數
    int inuse;      /* Offset to metadata */
    int align;      /* Alignment */
    int reserved;       /* Reserved bytes at the end of slabs */
    int red_left_pad;   /* Left redzone padding size */
    const char *name;   /* Name (only for display!) */
    struct list_head list;  /* List of slab caches */       //kmem_cache最終會鏈接在一個全局鏈表中
    struct kmem_cache_node *node[MAX_NUMNODES];     //Node管理slab頁面
};
  • struct kmem_cache_cpu：用於管理每個CPU的slab頁面，可以使用無鎖訪問，提高緩存對象分配速度；
struct kmem_cache_cpu {
    void **freelist;    /* Pointer to next available object */                  //指向空閑對象的指針
    unsigned long tid;  /* Globally unique transaction id */                
    struct page *page;  /* The slab from which we are allocating */     //slab緩存頁面
#ifdef CONFIG_SLUB_CPU_PARTIAL
    struct page *partial;   /* Partially allocated frozen slabs */
#endif
#ifdef CONFIG_SLUB_STATS
    unsigned stat[NR_SLUB_STAT_ITEMS];
#endif
};
  • struct kmem_cache_node：用於管理每個Node的slab頁面，由於每個Node的訪問速度不一致，slab頁面由Node來管理；
/*
 * The slab lists for all objects.
 */
struct kmem_cache_node {
    spinlock_t list_lock;

#ifdef CONFIG_SLUB
    unsigned long nr_partial;    //slab頁表數量
    struct list_head partial;       //slab頁面鏈表
#ifdef CONFIG_SLUB_DEBUG
    atomic_long_t nr_slabs;
    atomic_long_t total_objects;
    struct list_head full;
#endif
#endif
};
  • struct page：用於描述slab頁面，struct page結構體中很多字段都是通過union聯合體進行復用的。
    struct page結構中，用於slub的成員如下：
struct page {
    union {
       ...
        void *s_mem;            /* slab first object */
       ...
    };
    
    /* Second double word */
    union {
       ...
        void *freelist;     /* sl[aou]b first free object */
       ...
    };
    
    union {
       ...
        struct {
            union {
              ...
                struct {            /* SLUB */
                    unsigned inuse:16;
                    unsigned objects:15;
                    unsigned frozen:1;
                };
                ...
            };
       ...
        };       
    };   
    
    /*
     * Third double word block
     */
    union {
       ...
        struct {        /* slub per cpu partial pages */
            struct page *next;  /* Next partial slab */
#ifdef CONFIG_64BIT
            int pages;  /* Nr of partial slabs left */
            int pobjects;   /* Approximate # of objects */
#else
            short int pages;
            short int pobjects;
#endif
        };

        struct rcu_head rcu_head;   /* Used by SLAB
                         * when destroying via RCU
                         */
    };
    ...
        struct kmem_cache *slab_cache;  /* SL[AU]B: Pointer to slab */    
    ...
}

圖來了：

3. 流程分析

針對Slub的使用，可以從三個維度來分析：

  1. slub緩存創建
  2. slub對象分配
  3. slub對象釋放

下邊將進一步分析。

3.1 kmem_cache_create

在內核中通過kmem_cache_create接口來創建一個slab緩存。

先看一下這個接口的函數調用關係圖：

  1. kmem_cache_create完成的功能比較簡單，就是創建一個用於管理slab緩存的kmem_cache結構，並對該結構體進行初始化，最終添加到全局鏈表中。kmem_cache結構體初始化，包括了上文中分析到的kmem_cache_cpu和kmem_cache_node兩個字段結構。

  2. 在創建的過程中，當發現已有的slab緩存中，有存在對象大小相近，且具有兼容標誌的slab緩存，那就只需要進行merge操作並返回，而無需進一步創建新的slab緩存。

  3. calculate_sizes函數會根據指定的force_order或根據對象大小去計算kmem_cache結構體中的size/min/oo等值，其中kmem_cache_order_objects結構體，是由頁面分配order值和對象數量兩者通過位域拼接起來的。

  4. 在創建slab緩存的時候，有一個先雞后蛋的問題：kmem_cache結構體來管理一個slab緩存，而創建kmem_cache結構體又是從slab緩存中分配出來的對象，那麼這個問題是怎麼解決的呢？可以看一下kmem_cache_init函數，內核中定義了兩個靜態的全局變量kmem_cache和kmem_cache_node，在kmem_cache_init函數中完成了這兩個結構體的初始化之後，相當於就是創建了兩個slab緩存，一個用於分配kmem_cache結構體對象的緩存池，一個用於分配kmem_cache_node結構體對象的緩存池。由於kmem_cache_cpu結構體是通過__alloc_percpu來分配的，因此不需要創建一個相關的slab緩存。

3.2 kmem_cache_alloc

kmem_cache_alloc接口用於從slab緩存池中分配對象。

看一下大體的調用流程圖：

從上圖中可以看出，分配slab對象與Buddy System中分配頁面類似，存在快速路徑和慢速路徑兩種，所謂的快速路徑就是per-CPU緩存，可以無鎖訪問，因而效率更高。

整體的分配流程大體是這樣的：優先從per-CPU緩存中進行分配，如果per-CPU緩存中已經全部分配完畢，則從Node管理的slab頁面中遷移slab頁到per-CPU緩存中，再重新分配。當Node管理的slab頁面也不足的情況下，則從Buddy System中分配新的頁面，添加到per-CPU緩存中。

還是用圖來說明更清晰，分為以下幾步來分配：

  1. fastpath
    快速路徑下，以原子的方式檢索per-CPU緩存的freelist列表中的第一個對象，如果freelist為空並且沒有要檢索的對象，則跳入慢速路徑操作，最後再返回到快速路徑中重試操作。

  2. slowpath-1
    將per-CPU緩存中page指向的slab頁中的空閑對象遷移到freelist中，如果有空閑對象，則freeze該頁面，沒有空閑對象則跳轉到slowpath-2。

  3. slowpath-2
    將per-CPU緩存中partial鏈表中的第一個slab頁遷移到page指針中，如果partial鏈表為空，則跳轉到slowpath-3。

  4. slowpath-3
    將Node管理的partial鏈表中的slab頁遷移到per-CPU緩存中的page中，並重複第二個slab頁將其添加到per-CPU緩存中的partial鏈表中。如果遷移的slab中空閑對象超過了kmem_cache.cpu_partial的一半，則僅遷移slab頁，並且不再重複。
    如果每個Node的partial鏈表都為空，跳轉到slowpath-4。

  5. slowpath-4
    從Buddy System中獲取頁面，並將其添加到per-CPU的page中。

3.2 kmem_cache_free

kmem_cache_free的操作，可以看成是kmem_cache_alloc的逆過程，因此也分為快速路徑和慢速路徑兩種方式，同時，慢速路徑中又分為了好幾種情況，可以參考kmem_cache_alloc的過程。

調用流程圖如下：

效果如下：

  1. 快速路徑釋放
    快速路徑下，直接將對象返回到freelist中即可。

  2. put_cpu_partial
    put_cpu_partial函數主要是將一個剛freeze的slab頁，放入到partial鏈表中。
    在put_cpu_partial函數中調用unfreeze_partials函數，這時候會將per-CPU管理的partial鏈表中的slab頁面添加到Node管理的partial鏈表的尾部。如果超出了Node的partial鏈表，溢出的slab頁面中沒有分配對象的slab頁面將會返回到夥伴系統。

  3. add_partial
    添加slab頁到Node的partial鏈表中。

  4. remove_partial
    從Node的partial鏈表移除slab頁。

具體釋放的流程走哪個分支，跟對象的使用情況，partial鏈表的個數nr_partial/min_partial等相關，細節就不再深入分析了。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

.NET高級特性-Emit(1)

access_timePosted on by admin

　　在這個大數據/雲計算/人工智能研發普及的時代，Python的崛起以及Javascript的前後端的侵略，程序員與企業似乎越來越青睞動態語言所帶來的便捷性與高效性，即使靜態語言在性能，錯誤檢查等方面的優於靜態語言。對於.NETer來說，.NET做為一門靜態語言，我們不僅要打好.NET的基本功，如基本類型/語法/底層原理/錯誤檢查等知識，也要深入理解.NET的一些高級特性，來為你的工作減輕負擔和提高代碼質量。

　　ok,咱們今天開始聊一聊.NET中的Emit。

一、什麼是Emit？

　　Emit含義為發出、產生的含義，這是.NET中的一組類庫，命名空間為System.Reflection.Emit，幾乎所有的.NET版本(Framework/Mono/NetCore)都支持Emit,可以實現用C#代碼生成代碼的類庫

二、Emit的本質

　　我們知道.NET可以由各種語言進行編寫，比如VB,C++等，當然絕大部分程序員進行.NET開發都是使用C#語言進行的，這些語言都會被各自的語言解釋器解釋為IL語言並執行，而Emit類庫的作用就是用這些語言來編寫生成IL語言，並交給CLR(公共語言運行時)進行執行。

　　我們先來看看IL語言長什麼樣子：

　　(1) 首先我們創建一個Hello，World程序

    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Hello World!");
        }
    }

　　(2) 將程序編譯成dll文件，我們可以看到在開發目錄下生成了bin文件夾

　　

　　(3) 向下尋找，我們可以看到dll文件已經生成，筆者使用netcore3進行開發，故路徑為bin/Debug/netcoreapp3.0

　　

　　(4) 這時候，我們就要祭出我們的il查看神器了，ildasm工具

　　

　　如何找到這個工具？打開開始菜單，找到Visual Studio文件夾，打開Developer Command Prompt，在打開的命令行中鍵入ildasm回車即可，筆者使用vs2019進行演示，其它vs版本操作方法均一致

　　

 

 

 

 

 

 

 　　(5) 在dasm菜單欄選擇文件->打開，選擇剛剛生成的dll文件

　　

 

 

 　　(6) 即可查看生成il代碼

　　

 

　　有了ildasm的輔助，我們就能夠更好的了解IL語言以及如何編寫IL語言，此外，Visual Studio中還有許多插件支持查看il代碼，比如JetBrains出品的Resharper插件等，如果覺得筆者方式較為麻煩可以使用以上插件查看il代碼

三、理解IL代碼

　　在上一章節中，我們理解了Emit的本質其實就是用C#來編寫IL代碼，既然要編寫IL代碼，那麼我們首先要理解IL代碼是如何進行工作的，IL代碼是如何完成C#當中的順序/選擇/循環結構的，是如何實現類的定義/字段的定義/屬性的定義/方法的定義的。

　　IL代碼是一種近似於指令式的代碼語言，與彙編語言比較相近，所以習慣於寫高級語言的.NETer來說比較難以理解

　　讓我們來看看Hello，World程序的IL代碼:

IL_0000:  nop
IL_0001:  ldstr      "Hello World!"
IL_0006:  call       void [System.Console]System.Console::WriteLine(string)
IL_000b:  nop
IL_000c:  ret

　　我們可以把IL代碼看成棧的運行

　　第一條指令，nop表示不做任何事情，表示代碼不做任何事情

　　第二條指令，ldstr表示將字符串放入棧中，字符串的值為“Hello,World!”

　　第三條指令，call表示調用方法，參數為調用方法的方法信息，並把返回的結構壓入棧中，使用的參數為之前已經入棧的“Hello World!”，以此類推，如果方法有n個參數，那麼他就會調取棧中n個數據，並返回一個結果放回棧中

　　第四條指令，nop表示不做任何事情

　　第五條指令，ret表示將棧中頂部的數據返回，如果方法定義為void，則無返回值

　　關於Hello，world程序IL的理解就說到這裏，更多的指令含義讀者可以參考微軟官方文檔，筆者之後也會繼續對Emit進行講解和Emit的應用

四、用Emit類庫編寫IL代碼

　　既然IL代碼咱們理解的差不多了，咱們就開始嘗試用C#來寫IL代碼了，有了IL代碼的參考，咱們也可以依葫蘆畫瓢的把代碼寫出來了

　　(1) 引入Emit命名空間

using System.Reflection.Emit;

　　(2) 首先我們定義一個Main方法，入參無，返回類型void

//定義方法名，返回類型，輸入類型
var method = new DynamicMethod("Main", null, Type.EmptyTypes);

　　(3) 生成IL代碼

//生成IL代碼
var ilGenerator = method.GetILGenerator();
ilGenerator.Emit(OpCodes.Nop);
ilGenerator.Emit(OpCodes.Ldstr,"Hello World!");
ilGenerator.Emit(OpCodes.Call, typeof(Console).GetMethod("WriteLine", new Type[] { typeof(string) })); //尋找Console的WriteLine方法
ilGenerator.Emit(OpCodes.Nop);
ilGenerator.Emit(OpCodes.Ret);

　　(4) 創建委託並調用

//創建委託
var helloWorldMethod = method.CreateDelegate(typeof(Action)) as Action;
helloWorldMethod.Invoke();

　　(5)運行，即輸出Hello World!

五、小結

　　Emit的本質是使用高級語言生成IL代碼，進而進行調用的的一組類庫，依賴Emit我們可以實現用代碼生成代碼的操作，即編程語言的自舉，可以有效彌補靜態語言的靈活性的缺失。

　　Emit的性能非常好，除了第一次構建IL代碼所需要時間外，之後只要將操作緩存在計算機內存中，速度與手寫代碼相差無幾

　　有許多著名.NET類庫均依賴於Emit:

　　(.NET JSON操作庫)Json.NET/Newtonsoft.Json:

　　(輕量ORM)Dapper:

　　(ObjectToObjectMapper)EmitMapper:

　　(AOP庫)Castle.DynamicProxy:

　　學習Emit:

　　.NET官方文檔:

　　.NET API瀏覽器:

　　之後作者將繼續講解.NET Emit的相關內容和應用，感謝閱讀

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享

MySQL 複製表結構和表數據

access_timePosted on by admin

1、前言

　　在功能開發完畢，在本地或者測試環境進行測試時，經常會遇到這種情況：有專門的測試數據，測試過程會涉及到修改表中的數據，經常不能一次測試成功，所以，每次執行測試后，原來表中的數據其實已經被修改了，下一次測試，就需要將數據恢復。

　　我一般的做法是：先創建一個副本表，比如測試使用的user表，我在測試前創建副本表user_bak，每次測試后，將user表清空，然後將副本表user_bak的數據導入到user表中。

　　上面的操作是對一個table做備份，如果涉及到的table太多，可以創建database的副本。

　　接下來我將對此處的表結構複製以及表數據複製進行闡述，並非數據庫的複製原理！！！！

　　下面是staff表的表結構

create table staff (
	id int not null auto_increment comment '自增id',
	name char(20) not null comment '用戶姓名',
	dep char(20) not null comment '所屬部門',
	gender tinyint not null default 1 comment '性別：1男; 2女',
	addr char(30) not null comment '地址',
	primary key(id),
	index idx_1 (name, dep),
	index idx_2 (name, gender)
) engine=innodb default charset=utf8mb4 comment '員工表';

　

2、具體方式　

　　2.1、執行舊錶的創建SQL來創建表

　　如果原始表已經存在，那麼可以使用命令查看該表的創建語句：

mysql> show create table staff\G
*************************** 1. row ***************************
       Table: staff
Create Table: CREATE TABLE `staff` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` char(20) NOT NULL COMMENT '用戶姓名',
  `dep` char(20) NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_1` (`name`,`dep`),
  KEY `idx_2` (`name`,`gender`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='員工表'
1 row in set (0.01 sec)

　　可以看到，上面show creat table xx的命令執行結果中，Create Table的值就是創建表的語句，此時可以直接複製創建表的SQL，然後重新執行一次就行了。

　　當數據表中有數據的時候，看到的創建staff表的sql就會稍有不同。比如，我在staff中添加了兩條記錄：

mysql> insert into staff values (null, '李明', 'RD', 1, '北京');
Query OK, 1 row affected (0.00 sec)

mysql> insert into staff values (null, '張三', 'PM', 0, '上海');
Query OK, 1 row affected (0.00 sec)

mysql> select * from staff;
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

　　此時在執行show create table命令：

mysql> show create table staff\G
*************************** 1. row ***************************
       Table: staff
Create Table: CREATE TABLE `staff` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` char(20) NOT NULL COMMENT '用戶姓名',
  `dep` char(20) NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_1` (`name`,`dep`),
  KEY `idx_2` (`name`,`gender`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT='員工表'
1 row in set (0.00 sec)

　　注意，上面結果中的倒數第二行

　　　　ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT=’員工表’

　　因為staff表的id是自增的，且已經有了2條記錄，所以下一次插入數據的自增id應該為3，這個信息，也會出現在表的創建sql中。

　　　　

　　2.2、使用like創建新表（僅包含表結構）

　　使用like根據已有的表來創建新表，特點如下：

　　1、方便，不需要查看原表的表結構定義信息；

　　2、創建的新表中，表結構定義、完整性約束，都與原表保持一致。

　　3、創建的新表是一個空表，全新的表，沒有數據。

　　用法如下：

mysql> select * from staff;　　#舊錶中已有2條數據
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

mysql> create table staff_bak_1 like staff;  # 直接使用like，前面指定新表名，後面指定舊錶（參考的表）
Query OK, 0 rows affected (0.02 sec)

mysql> show create table staff_bak_1\G
*************************** 1. row ***************************
       Table: staff_bak_1
Create Table: CREATE TABLE `staff_bak_1` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` char(20) NOT NULL COMMENT '用戶姓名',
  `dep` char(20) NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `idx_1` (`name`,`dep`),
  KEY `idx_2` (`name`,`gender`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='員工表'  # 注意沒有AUTO_INCREMENT=3
1 row in set (0.00 sec)

mysql> select * from staff_bak_1; # 沒有包含舊錶的數據
Empty set (0.00 sec)

　　

　　2.3、使用as來創建新表（包含數據）

　　使用as來創建新表，有一下特點：

　　1、可以有選擇性的決定新表包含哪些字段；

　　2、創建的新表中，會包含舊錶的數據；

　　3、創建的新表不會包含舊錶的完整性約束（比如主鍵、索引等），僅包含最基礎的表結構定義。

　　用法如下：

mysql> create table staff_bak_2 as select * from staff;
Query OK, 2 rows affected (0.02 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from staff_bak_2;
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

mysql> show create table staff_bak_2\G
*************************** 1. row ***************************
       Table: staff_bak_2
Create Table: CREATE TABLE `staff_bak_2` (
  `id` int(11) NOT NULL DEFAULT '0' COMMENT '自增id',
  `name` char(20) CHARACTER SET utf8mb4 NOT NULL COMMENT '用戶姓名',
  `dep` char(20) CHARACTER SET utf8mb4 NOT NULL COMMENT '所屬部門',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) CHARACTER SET utf8mb4 NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1
1 row in set (0.00 sec)

　　

　　利用as創建表的時候沒有保留完整性約束，其實這個仔細想一下也能想明白。因為使用as創建表的時候，可以指定新表包含哪些字段呀，如果你創建新表時，忽略了幾個字段，這樣的話即使保留了完整約束，保存數據是也不能滿足完整性約束。

　　比如，staff表有一個索引idx1，由name和dep字段組成；但是我創建的新表中，沒有name和dep字段（只選擇了其他字段），那麼新表中保存idx1也沒有必要，對吧。

mysql> --  只選擇id、gender、addr作為新表的字段，那麼name和dep組成的索引就沒必要存在了
mysql> create table staff_bak_3 as (select id, gender, addr from staff);
Query OK, 2 rows affected (0.02 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> show create table staff_bak_3\G
*************************** 1. row ***************************
       Table: staff_bak_3
Create Table: CREATE TABLE `staff_bak_3` (
  `id` int(11) NOT NULL DEFAULT '0' COMMENT '自增id',
  `gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `addr` char(30) CHARACTER SET utf8mb4 NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1
1 row in set (0.00 sec)

mysql> select * from staff_bak_3;
+----+--------+--------+
| id | gender | addr   |
+----+--------+--------+
|  1 |      1 | 北京   |
|  2 |      0 | 上海   |
+----+--------+--------+
2 rows in set (0.00 sec)

　　

　　2.4、使用like+insert+select創建原表的副本（推薦）

　　使用like創建新表，雖然保留了舊錶的各種表結構定義以及完整性約束，但是如何將舊錶的數據導入到新表中呢？

　　最極端的方式：寫一個程序，先將舊錶數據讀出來，然後寫入到新表中，這個方式我就不嘗試了。

　　有一個比較簡單的命令：

mysql> select * from staff; #原表數據
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

mysql> select * from staff_bak_1; # 使用like創建的表，與原表相同的表結構和完整性約束（自增除外）
Empty set (0.00 sec)

mysql> insert into staff_bak_1 select * from staff;  # 將staff表的所有記錄的所有字段值都插入副本表中
Query OK, 2 rows affected (0.00 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from staff_bak_1;
+----+--------+-----+--------+--------+
| id | name   | dep | gender | addr   |
+----+--------+-----+--------+--------+
|  1 | 李明   | RD  |      1 | 北京   |
|  2 | 張三   | PM  |      0 | 上海   |
+----+--------+-----+--------+--------+
2 rows in set (0.00 sec)

　　

　　其實這條SQL語句，是知道兩個表的表結構和完整性約束相同，所以，可以直接select *。

insert into staff_bak_1 select * from staff;

　　

　　如果兩個表結構不相同，其實也是可以這個方式的，比如：

mysql> show create table demo\G
*************************** 1. row ***************************
       Table: demo
Create Table: CREATE TABLE `demo` (
  `_id` int(11) NOT NULL AUTO_INCREMENT,
  `_name` char(20) DEFAULT NULL,
  `_gender` tinyint(4) DEFAULT '1',
  PRIMARY KEY (`_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
1 row in set (0.00 sec)

# 只將staff表中的id和name字段組成的數據記錄插入到demo表中，對應_id和_name字段
mysql> insert into demo (_id, _name) select id,name from staff;
Query OK, 2 rows affected (0.00 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from demo;
+-----+--------+---------+
| _id | _name  | _gender |
+-----+--------+---------+
|   1 | 李明   |       1 |
|   2 | 張三   |       1 |
+-----+--------+---------+
2 rows in set (0.00 sec)

　　這是兩個表的字段數量不相同的情況，此時需要手動指定列名，否則就會報錯。

 

　　另外，如果兩個表的字段數量，以及相同順序的字段類型相同，如果是全部字段複製，即使字段名不同，也可以直接複製：

# staff_bak_5的字段名與staff表並不相同，但是字段數量、相同順序字段的類型相同，所以可以直接插入
mysql> show create table staff_bak_5\G
*************************** 1. row ***************************
       Table: staff_bak_5
Create Table: CREATE TABLE `staff_bak_5` (
  `_id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `_name` char(20) NOT NULL COMMENT '用戶姓名',
  `_dep` char(20) NOT NULL COMMENT '所屬部門',
  `_gender` tinyint(4) NOT NULL DEFAULT '1' COMMENT '性別：1男; 2女',
  `_addr` char(30) NOT NULL,
  PRIMARY KEY (`_id`),
  KEY `idx_1` (`_name`,`_dep`),
  KEY `idx_2` (`_name`,`_gender`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COMMENT='員工表'
1 row in set (0.00 sec)

mysql> insert into staff_bak_5 select * from staff;
Query OK, 2 rows affected (0.00 sec)
Records: 2  Duplicates: 0  Warnings: 0

mysql> select * from staff_bak_5;
+-----+--------+------+---------+--------+
| _id | _name  | _dep | _gender | _addr  |
+-----+--------+------+---------+--------+
|   1 | 李明   | RD   |       1 | 北京   |
|   2 | 張三   | PM   |       0 | 上海   |
+-----+--------+------+---------+--------+
2 rows in set (0.00 sec)

　　

　　

 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※台北網頁設計公司這麼多，該如何挑選?? 網頁設計報價省錢懶人包"嚨底家"

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※想知道購買電動車哪裡補助最多?台中電動車補助資訊懶人包彙整

源碼分析RocketMQ消息軌跡

access_timePosted on by admin

目錄

本文沿着的思路，從如下3個方面對其源碼進行解讀：

  1. 發送消息軌跡
  2. 消息軌跡格式
  3. 存儲消息軌跡數據

@(本節目錄)

1、發送消息軌跡流程

首先我們來看一下在消息發送端如何啟用消息軌跡，示例代碼如下：

public class TraceProducer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true);      // @1
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        for (int i = 0; i < 10; i++)
            try {
                {
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                }

            } catch (Exception e) {
                e.printStackTrace();
            }
        producer.shutdown();
    }
}

從上述代碼可以看出其關鍵點是在創建DefaultMQProducer時指定開啟消息軌跡跟蹤。我們不妨瀏覽一下DefaultMQProducer與啟用消息軌跡相關的構造函數：

public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace)
public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace, final String customizedTraceTopic)

參數如下：

  • String producerGroup
    生產者所屬組名。
  • boolean enableMsgTrace
    是否開啟跟蹤消息軌跡，默認為false。
  • String customizedTraceTopic
    如果開啟消息軌跡跟蹤，用來存儲消息軌跡數據所屬的主題名稱，默認為：RMQ_SYS_TRACE_TOPIC。

1.1 DefaultMQProducer構造函數

public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic) {      // @1
    this.producerGroup = producerGroup;
    defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
    //if client open the message trace feature
    if (enableMsgTrace) {                                                                                                                                                                                            // @2
        try {
            AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(customizedTraceTopic, rpcHook);                                                         
            dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
            traceDispatcher = dispatcher;
            this.getDefaultMQProducerImpl().registerSendMessageHook(
                new SendMessageTraceHookImpl(traceDispatcher));                                                                                                                             // @3
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
        }
    }
}

代碼@1：首先介紹一下其局部變量。

  • String producerGroup
    生產者所屬組。
  • RPCHook rpcHook
    生產者發送鈎子函數。
  • boolean enableMsgTrace
    是否開啟消息軌跡跟蹤。
  • String customizedTraceTopic
    定製用於存儲消息軌跡的數據。

代碼@2：用來構建AsyncTraceDispatcher，看其名：異步轉發消息軌跡數據，稍後重點關注。

代碼@3：構建SendMessageTraceHookImpl對象，並使用AsyncTraceDispatcher用來異步轉發。

1.2 SendMessageTraceHookImpl鈎子函數

1.2.1 SendMessageTraceHookImpl類圖

  1. SendMessageHook
    消息發送鈎子函數，用於在消息發送之前、發送之後執行一定的業務邏輯，是記錄消息軌跡的最佳擴展點。
  2. TraceDispatcher
    消息軌跡轉發處理器，其默認實現類AsyncTraceDispatcher，異步實現消息軌跡數據的發送。下面對其屬性做一個簡單的介紹：
    • int queueSize
      異步轉發，隊列長度，默認為2048，當前版本不能修改。
    • int batchSize
      批量消息條數，消息軌跡一次消息發送請求包含的數據條數，默認為100，當前版本不能修改。
    • int maxMsgSize
      消息軌跡一次發送的最大消息大小，默認為128K，當前版本不能修改。
    • DefaultMQProducer traceProducer
      用來發送消息軌跡的消息發送者。
    • ThreadPoolExecutor traceExecuter
      線程池，用來異步執行消息發送。
    • AtomicLong discardCount
      記錄丟棄的消息個數。
    • Thread worker
      woker線程，主要負責從追加隊列中獲取一批待發送的消息軌跡數據，提交到線程池中執行。
    • ArrayBlockingQueue< TraceContext> traceContextQueue
      消息軌跡TraceContext隊列，用來存放待發送到服務端的消息。
    • ArrayBlockingQueue< Runnable> appenderQueue
      線程池內部隊列，默認長度1024。
    • DefaultMQPushConsumerImpl hostConsumer
      消費者信息，記錄消息消費時的軌跡信息。
    • String traceTopicName
      用於跟蹤消息軌跡的topic名稱。

1.2.2 源碼分析SendMessageTraceHookImpl

1.2.2.1 sendMessageBefore
public void sendMessageBefore(SendMessageContext context) { 
    //if it is message trace data,then it doesn't recorded
    if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) {   // @1
        return;
    }
    //build the context content of TuxeTraceContext
    TraceContext tuxeContext = new TraceContext();
    tuxeContext.setTraceBeans(new ArrayList<TraceBean>(1));
    context.setMqTraceContext(tuxeContext);
    tuxeContext.setTraceType(TraceType.Pub);
    tuxeContext.setGroupName(context.getProducerGroup());                                                                                                                       // @2
    //build the data bean object of message trace
    TraceBean traceBean = new TraceBean();                                                                                                                                                // @3
    traceBean.setTopic(context.getMessage().getTopic());
    traceBean.setTags(context.getMessage().getTags());
    traceBean.setKeys(context.getMessage().getKeys());
    traceBean.setStoreHost(context.getBrokerAddr());
    traceBean.setBodyLength(context.getMessage().getBody().length);
    traceBean.setMsgType(context.getMsgType());
    tuxeContext.getTraceBeans().add(traceBean);
}

代碼@1：如果topic主題為消息軌跡的Topic，直接返回。

代碼@2：在消息發送上下文中，設置用來跟蹤消息軌跡的上下環境，裏面主要包含一個TraceBean集合、追蹤類型（TraceType.Pub）與生產者所屬的組。

代碼@3：構建一條跟蹤消息，用TraceBean來表示，記錄原消息的topic、tags、keys、發送到broker地址、消息體長度等消息。

從上文看出，sendMessageBefore主要的用途就是在消息發送的時候，先準備一部分消息跟蹤日誌，存儲在發送上下文環境中，此時並不會發送消息軌跡數據。

1.2.2.2 sendMessageAfter
public void sendMessageAfter(SendMessageContext context) {
    //if it is message trace data,then it doesn't recorded
    if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())     // @1
        || context.getMqTraceContext() == null) {
        return;
    }
    if (context.getSendResult() == null) {
        return;
    }

    if (context.getSendResult().getRegionId() == null
        || !context.getSendResult().isTraceOn()) {
        // if switch is false,skip it
        return;
    }

    TraceContext tuxeContext = (TraceContext) context.getMqTraceContext();
    TraceBean traceBean = tuxeContext.getTraceBeans().get(0);                                                                                                // @2
    int costTime = (int) ((System.currentTimeMillis() - tuxeContext.getTimeStamp()) / tuxeContext.getTraceBeans().size());     // @3
    tuxeContext.setCostTime(costTime);                                                                                                                                      // @4
    if (context.getSendResult().getSendStatus().equals(SendStatus.SEND_OK)) {                                                                    
        tuxeContext.setSuccess(true);
    } else {
        tuxeContext.setSuccess(false);
    }
    tuxeContext.setRegionId(context.getSendResult().getRegionId());                                                                                      
    traceBean.setMsgId(context.getSendResult().getMsgId());
    traceBean.setOffsetMsgId(context.getSendResult().getOffsetMsgId());
    traceBean.setStoreTime(tuxeContext.getTimeStamp() + costTime / 2);
    localDispatcher.append(tuxeContext);                                                                                                                                   // @5
}

代碼@1：如果topic主題為消息軌跡的Topic，直接返回。

代碼@2：從MqTraceContext中獲取跟蹤的TraceBean，雖然設計成List結構體，但在消息發送場景，這裏的數據永遠只有一條，及時是批量發送也不例外。

代碼@3：獲取消息發送到收到響應結果的耗時。

代碼@4：設置costTime(耗時)、success(是否發送成功)、regionId(發送到broker所在的分區)、msgId(消息ID，全局唯一)、offsetMsgId(消息物理偏移量，如果是批量消息，則是最後一條消息的物理偏移量)、storeTime，這裏使用的是(客戶端發送時間 + 二分之一的耗時)來表示消息的存儲時間，這裡是一個估值。

代碼@5：將需要跟蹤的信息通過TraceDispatcher轉發到Broker服務器。其代碼如下：

public boolean append(final Object ctx) {
    boolean result = traceContextQueue.offer((TraceContext) ctx);
    if (!result) {
        log.info("buffer full" + discardCount.incrementAndGet() + " ,context is " + ctx);
    }
    return result;
}

這裏一個非常關鍵的點是offer方法的使用，當隊列無法容納新的元素時會立即返回false，並不會阻塞。

接下來將目光轉向TraceDispatcher的實現。

1.3 TraceDispatcher實現原理

TraceDispatcher，用於客戶端消息軌跡數據轉發到Broker，其默認實現類：AsyncTraceDispatcher。

1.3.1 TraceDispatcher構造函數

public AsyncTraceDispatcher(String traceTopicName, RPCHook rpcHook) throws MQClientException {    
    // queueSize is greater than or equal to the n power of 2 of value
    this.queueSize = 2048;
    this.batchSize = 100;
    this.maxMsgSize = 128000;                                        
    this.discardCount = new AtomicLong(0L);         
    this.traceContextQueue = new ArrayBlockingQueue<TraceContext>(1024);
    this.appenderQueue = new ArrayBlockingQueue<Runnable>(queueSize);
    if (!UtilAll.isBlank(traceTopicName)) {
        this.traceTopicName = traceTopicName;
    } else {
        this.traceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
    }                   // @1
    this.traceExecuter = new ThreadPoolExecutor(// :
        10, //
        20, //
        1000 * 60, //
        TimeUnit.MILLISECONDS, //
        this.appenderQueue, //
        new ThreadFactoryImpl("MQTraceSendThread_"));
    traceProducer = getAndCreateTraceProducer(rpcHook);      // @2
}

代碼@1：初始化核心屬性，該版本這些值都是“固化”的，用戶無法修改。

  • queueSize
    隊列長度，默認為2048，異步線程池能夠積壓的消息軌跡數量。
  • batchSize
    一次向Broker批量發送的消息條數，默認為100.
  • maxMsgSize
    向Broker彙報消息軌跡時，消息體的總大小不能超過該值，默認為128k。
  • discardCount
    整個運行過程中，丟棄的消息軌跡數據，這裏要說明一點的是，如果消息TPS發送過大，異步轉發線程處理不過來時，會主動丟棄消息軌跡數據。
  • traceContextQueue
    traceContext積壓隊列，客戶端(消息發送、消息消費者)在收到處理結果后，將消息軌跡提交到噶隊列中，則會立即返回。
  • appenderQueue
    提交到Broker線程池中隊列。
  • traceTopicName
    用於接收消息軌跡的Topic，默認為RMQ_SYS_TRANS_HALF_TOPIC。
  • traceExecuter
    用於發送到Broker服務的異步線程池，核心線程數默認為10，最大線程池為20，隊列堆積長度2048，線程名稱：MQTraceSendThread_。、
  • traceProducer
    發送消息軌跡的Producer。

代碼@2：調用getAndCreateTraceProducer方法創建用於發送消息軌跡的Producer(消息發送者)，下面詳細介紹一下其實現。

1.3.2 getAndCreateTraceProducer詳解

private DefaultMQProducer getAndCreateTraceProducer(RPCHook rpcHook) {
        DefaultMQProducer traceProducerInstance = this.traceProducer;
        if (traceProducerInstance == null) {  //@1
            traceProducerInstance = new DefaultMQProducer(rpcHook);
            traceProducerInstance.setProducerGroup(TraceConstants.GROUP_NAME);
            traceProducerInstance.setSendMsgTimeout(5000);
            traceProducerInstance.setVipChannelEnabled(false);
            // The max size of message is 128K
            traceProducerInstance.setMaxMessageSize(maxMsgSize - 10 * 1000);
        }
        return traceProducerInstance;
    }

代碼@1：如果還未建立發送者，則創建用於發送消息軌跡的消息發送者，其GroupName為：_INNER_TRACE_PRODUCER，消息發送超時時間5s，最大允許發送消息大小118K。

1.3.3 start

public void start(String nameSrvAddr) throws MQClientException {
    if (isStarted.compareAndSet(false, true)) {     // @1
        traceProducer.setNamesrvAddr(nameSrvAddr);
        traceProducer.setInstanceName(TRACE_INSTANCE_NAME + "_" + nameSrvAddr);
        traceProducer.start();
    }
    this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);   // @2
    this.worker.setDaemon(true);
    this.worker.start();                                                                                   
    this.registerShutDownHook();
}

開始啟動，其調用的時機為啟動DefaultMQProducer時，如果啟用跟蹤消息軌跡，則調用之。

代碼@1：如果用於發送消息軌跡的發送者沒有啟動，則設置nameserver地址，並啟動着。

代碼@2：啟動一個線程，用於執行AsyncRunnable任務，接下來將重點介紹。

1.3.4 AsyncRunnable

class AsyncRunnable implements Runnable {
         private boolean stopped;
    public void run() {
        while (!stopped) {
            List<TraceContext> contexts = new ArrayList<TraceContext>(batchSize);     // @1
            for (int i = 0; i < batchSize; i++) {
                TraceContext context = null;
                try {
                    //get trace data element from blocking Queue — traceContextQueue
                    context = traceContextQueue.poll(5, TimeUnit.MILLISECONDS);        // @2
                } catch (InterruptedException e) {
                }
                if (context != null) {
                    contexts.add(context);
                } else {
                    break;
                }
            }
            if (contexts.size() > 0) {                                                                               :
                AsyncAppenderRequest request = new AsyncAppenderRequest(contexts);  // @3
                traceExecuter.submit(request);                                                               
            } else if (AsyncTraceDispatcher.this.stopped) {
                this.stopped = true;
            }
        }
    }
}

代碼@1：構建待提交消息跟蹤Bean，每次最多發送batchSize，默認為100條。

代碼@2：從traceContextQueue中取出一個待提交的TraceContext，設置超時時間為5s，即如何該隊列中沒有待提交的TraceContext，則最多等待5s。

代碼@3：向線程池中提交任務AsyncAppenderRequest。

1.3.5 AsyncAppenderRequest#sendTraceData

public void sendTraceData(List<TraceContext> contextList) {
    Map<String, List<TraceTransferBean>> transBeanMap = new HashMap<String, List<TraceTransferBean>>();
    for (TraceContext context : contextList) {        //@1
        if (context.getTraceBeans().isEmpty()) {
            continue;
        }
        // Topic value corresponding to original message entity content
        String topic = context.getTraceBeans().get(0).getTopic();     // @2
        // Use  original message entity's topic as key
        String key = topic;
        List<TraceTransferBean> transBeanList = transBeanMap.get(key);
        if (transBeanList == null) {
            transBeanList = new ArrayList<TraceTransferBean>();
            transBeanMap.put(key, transBeanList);
        }
        TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context);    // @3
        transBeanList.add(traceData);
    }
    for (Map.Entry<String, List<TraceTransferBean>> entry : transBeanMap.entrySet()) {       // @4
        flushData(entry.getValue());
    }
}

代碼@1：遍歷收集的消息軌跡數據。

代碼@2：獲取存儲消息軌跡的Topic。

代碼@3：對TraceContext進行編碼，這裡是消息軌跡的傳輸數據，稍後對其詳細看一下，了解其上傳的格式。

代碼@4：將編碼后的數據發送到Broker服務器。

1.3.6 TraceDataEncoder#encoderFromContextBean

根據消息軌跡跟蹤類型，其格式會有一些不一樣，下面分別來介紹其合適。

1.3.6.1 PUB(消息發送)
case Pub: {
    TraceBean bean = ctx.getTraceBeans().get(0);
    //append the content of context and traceBean to transferBean's TransData
    sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getTopic()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getTags()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getStoreHost()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getBodyLength()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getMsgType().ordinal()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getOffsetMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
     .append(ctx.isSuccess()).append(TraceConstants.FIELD_SPLITOR);
}

消息軌跡數據的協議使用字符串拼接，字段的分隔符號為1，整個數據以2結尾，感覺這個設計還是有點“不可思議”，為什麼不直接使用json協議呢？

1.3.6.2 SubBefore(消息消費之前)
for (TraceBean bean : ctx.getTraceBeans()) {
    sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getRetryTimes()).append(TraceConstants.CONTENT_SPLITOR)//
      .append(bean.getKeys()).append(TraceConstants.FIELD_SPLITOR);//
    }
}

軌跡就是按照上述順序拼接而成，各個字段使用1分隔，每一條記錄使用2結尾。

1.3.2.3 SubAfter（消息消費后）
case SubAfter: {
    for (TraceBean bean : ctx.getTraceBeans()) {
        sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.isSuccess()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
          .append(ctx.getContextCode()).append(TraceConstants.FIELD_SPLITOR);
        }
    }
}

格式編碼一樣，就不重複多說。

經過上面的源碼跟蹤，消息發送端的消息軌跡跟蹤流程、消息軌跡數據編碼協議就清晰了，接下來我們使用一張序列圖來結束本部分的講解。

其實行文至此，只關注了消息發送的消息軌跡跟蹤，消息消費的軌跡跟蹤又是如何呢？其實現原理其實是一樣的，就是在消息消費前後執行特定的鈎子函數，其實現類為ConsumeMessageTraceHookImpl，由於其實現與消息發送的思路類似，故就不詳細介紹了。

2、 消息軌跡數據如何存儲

其實從上面的分析，我們已經得知，RocketMQ的消息軌跡數據存儲在到Broker上，那消息軌跡的主題名如何指定？其路由信息又怎麼分配才好呢？是每台Broker上都創建還是只在其中某台上創建呢？RocketMQ支持系統默認與自定義消息軌跡的主題。

2.1 使用系統默認的主題名稱

RocketMQ默認的消息軌跡主題為：RMQ_SYS_TRACE_TOPIC，那該Topic需要手工創建嗎？其路由信息呢？

{
    if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {    // @1
        String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
        TopicConfig topicConfig = new TopicConfig(topic);
        this.systemTopicList.add(topic);
        topicConfig.setReadQueueNums(1);                                              // @2
        topicConfig.setWriteQueueNums(1);
        this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
    }
}

上述代碼出自TopicConfigManager的構造函數，在Broker啟動的時候會創建topicConfigManager對象，用來管理topic的路由信息。

代碼@1：如果Broker開啟了消息軌跡跟蹤(traceTopicEnable=true)時，會自動創建默認消息軌跡的topic路由信息，注意其讀寫隊列數為1。

2.2 用戶自定義消息軌跡主題

在創建消息發送者、消息消費者時，可以显示的指定消息軌跡的Topic，例如：

public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic)

public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
        AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean enableMsgTrace, final String customizedTraceTopic)

通過customizedTraceTopic來指定消息軌跡Topic。

溫馨提示：通常在生產環境上，將不會開啟自動創建主題，故需要RocketMQ運維管理人員提前創建好Topic。

好了，本文就介紹到這裏了，本文詳細介紹了RocktMQ消息軌跡的實現原理，下一篇，我們將進入到多副本的學習中。

作者介紹：
丁威，《RocketMQ技術內幕》作者，RocketMQ 社區佈道師，公眾號： 維護者，目前已陸續發表源碼分析Java集合、Java 併發包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源碼專欄。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※想知道網站建置、網站改版該如何進行嗎?將由專業工程師為您規劃客製化網頁設計及後台網頁設計

※不管是台北網頁設計公司、台中網頁設計公司，全省皆有專員為您服務

※Google地圖已可更新顯示潭子電動車充電站設置地點!!

※帶您來看台北網站建置，台北網頁設計，各種案例分享