標籤: 新北清潔

FreeSql 使用 ToTreeList/AsTreeCte 查詢無限級分類表

admin

關於無限級分類

第一種方案：
使用遞歸算法，也是使用頻率最多的，大部分開源程序也是這麼處理，不過一般都只用到四級分類。 這種算法的數據庫結構設計最為簡單。category表中一個字段id，一個字段fid(父id)。這樣可以根據WHERE id = fid來判斷上一級內容，運用遞歸至最頂層。
分析：通過這種數據庫設計出的無限級，可以說讀取的時候相當費勁，所以大部分的程序最多3-4級分類，這就足以滿足需求，從而一次性讀出所有的數據，再對得到數組或者對象進行遞歸。本身負荷還是沒太大問題。但是如果分類到更多級，那是不可取的辦法。
這樣看來這種分類有個好處，就是增刪改的時候輕鬆了…然而就二級分類而言，採用這種算法就應該算最優先了。

第二種方案：
設置fid字段類型為varchar，將父類id都集中在這個字段里，用符號隔開，比如：1,3,6
這樣可以比較容易得到各上級分類的ID，而且在查詢分類下的信息的時候，
可以使用：SELECT * FROM category WHERE pid LIKE “1,3%”。

分 析：相比於遞歸算法，在讀取數據方面優勢非常大，但是若查找該分類的所有 父分類 或者 子分類 查詢的效率也不是很高，至少也要二次query，從某種意義看上，個人覺得不太符合數據庫範式的設計。倘若遞增到無限級，還需考慮字段是否達到要求，而且 在修改分類和轉移分類的時候操作將非常麻煩。
暫時，在自己項目中用的就是類似第二種方案的解決辦法。就該方案在我的項目中存在這樣的問題， 如果當所有數據記錄達到上萬甚至10W以上后，一次性將所以分類，有序分級的現實出來，效率很低。極有可能是項目處理數據代碼效率低帶來的。現在正在改良。

第三種方案：
　　無限級分類—-改進前序遍歷樹
那 么理想中的樹型結構應具備哪些特點呢？數據存儲冗餘小、直觀性強；方便返回整個樹型結構數據；可以很輕鬆的返回某一子樹（方便分層加載）；快整獲以某節點 的祖譜路徑；插入、刪除、移動節點效率高等等。帶着這些需求我查找了很多資料，發現了一種理想的樹型結構數據存儲及操作算法，改進的前序遍歷樹模型 （The Nested Set Model）。
原理：
我們先把樹按照水平方式擺開。從根節點開始（“Food”），然後他的左邊寫 上1。然後按照樹的順序（從上到下）給“Fruit”的左邊寫上2。這樣，你沿着樹的邊界走啊走（這就是“遍歷”），然後同時在每個節點的左邊和右邊寫上 数字。最後，我們回到了根節點“Food”在右邊寫上18。下面是標上了数字的樹，同時把遍歷的順序用箭頭標出來了。

我 們稱這些数字為左值和右值（如，“Food”的左值是1，右值是18）。正如你所見，這些数字按時了每個節點之間的關係。因為“Red”有3和6兩個值， 所以，它是有擁有1-18值的“Food”節點的後續。同樣的，我們可以推斷所有左值大於2並且右值小於11的節點，都是有2-11的“Fruit” 節點的後續。這樣，樹的結構就通過左值和右值儲存下來了。這種數遍整棵樹算節點的方法叫做“改進前序遍歷樹”算法。

表結構設計：

那 么我們怎樣才能通過一個SQL語句把所有的分類都查詢出來呢，而且要求如果是子類的話前面要打幾個空格以表現是子分類。要想查詢出所有分類很好 辦：SELECT * FROM category WHERE lft>1 AND lft<18 ORDER BY lft這樣的話所有的分類都出來了，但是誰是誰的子類卻分不清，那麼怎麼辦呢?我們仔細看圖不難發現如果相鄰的兩條記錄的右值第一條的右值比第二條的大那 么就是他的父類，比如food的右值是18而fruit的右值是11 那麼food是fruit的父類，但是又要考慮到多級目錄。於是有了這樣的設計，我們用一個數組來存儲上一條記錄的右值，再把它和本條記錄的右值比較，如 果前者比後者小，說明不是父子關係，就用array_pop彈出數組，否則就保留，之後根據數組的大小來打印空格。

以上內容引用出處：https://www.cnblogs.com/badboys/p/9945296.html

關於第三種設計的更多資料請點擊查看原文，因為過於複雜（過重）被使用的頻率不高。

引出痛點

無限級分類（父子）是一種比較常用的表設計，每種設計方式突出優勢的同時也帶來缺陷，如：

  • 第一種方案：表設計中只有 parent_id 字段，寫入數據方便，困擾：查詢麻煩，許多使用了 ORM 的項目被迫使用 SQL 解決該場景；
  • 第二種方案：表設計中冗餘子級id便於查詢，困擾：添加/更新/刪除的時候需要重新計算；
  • 第三種方案：表設計中存儲左右值編碼，困擾：同上；

第一種方案的設計最簡單，本文後面的內容是在該基礎上，使用 FreeSql 實現 ToTreeList(內存加工樹型)、AsTreeCte(實現遞歸向下/向上查詢)，滿足大眾日常使用。

關於 FreeSql

FreeSql 是功能強大的對象關係映射技術(O/RM)，支持 .NETCore 2.1+ 或 .NETFramework 4.0+ 或 Xamarin，以 MIT 開源協議託管於 github，單元測試數量 4528個，nuget 下載量 151K，支持 MySql/SqlServer/PostgreSQL/Oracle/Sqlite/達夢/人大金倉/神州通用/Access；

源碼地址：https://github.com/dotnetcore/FreeSql

作者說過：每一個功能代表他的一撮頭髮！

第一步：定義導航屬性

FreeSql 導航屬性之中，有針對父子關係的設置方式，ToTreeList/AsTreeCte 依賴該設置，如下：

public class Area
{
  [Column(IsPrimary = true)]
  public string Code { get; set; }

  public string Name { get; set; }
  public string ParentCode { get; set; }

  [Navigate(nameof(ParentCode)), JsonIgnore] //JsonIgnore 是 json.net 的特性
  public Area Parent { get; set; }
  [Navigate(nameof(ParentCode))]
  public List<Area> Childs { get; set; }
}

關於導航屬性

定義 Parent 屬性，在表達式中可以這樣：

fsql.Select<Area>()
  .Where(a => a.Parent.Parent.Parent.Name == "中國")
  .First();

定義 Childs 屬性，在表達式中可以這樣（子查詢）：

fsql.Select<Area>()
  .Where(a => a.Childs.AsSelect().Any(c => c.Name == "北京"))
  .First();

定義 Childs 屬性，還可以使用【級聯保存】、【貪婪加載】 等等操作。

添加測試數據

fsql.Delete<Area>().Where("1=1").ExecuteAffrows();
var repo = fsql.GetRepository<Area>();
repo.DbContextOptions.EnableAddOrUpdateNavigateList = true;
repo.DbContextOptions.NoneParameter = true;
repo.Insert(new Area
{
  Code = "100000",
  Name = "中國",
  Childs = new List<Area>(new[] {
    new Area
    {
      Code = "110000",
      Name = "北京",
      Childs = new List<Area>(new[] {
        new Area{ Code="110100", Name = "北京市" },
        new Area{ Code="110101", Name = "東城區" },
      })
    }
  })
});

第二步：使用 ToTreeList 返回樹型數據

配置好父子屬性之後，就可以這樣用了：

var t1 = fsql.Select<Area>().ToTreeList();
Assert.Single(t1);
Assert.Equal("100000", t1[0].Code);
Assert.Single(t1[0].Childs);
Assert.Equal("110000", t1[0].Childs[0].Code);
Assert.Equal(2, t1[0].Childs[0].Childs.Count);
Assert.Equal("110100", t1[0].Childs[0].Childs[0].Code);
Assert.Equal("110101", t1[0].Childs[0].Childs[1].Code);

查詢數據本來是平面的，ToTreeList 方法將返回的平面數據在內存中加工為樹型 List 返回。

[
  {
    "ParentCode": null,
    "Childs": [
      {
        "ParentCode": "100000",
        "Childs": [
          {
            "ParentCode": "110000",
            "Childs": [],
            "Code": "110100",
            "Name": "北京市"
          },
          {
            "ParentCode": "110000",
            "Childs": [],
            "Code": "110101",
            "Name": "東城區"
          }
        ],
        "Code": "110000",
        "Name": "北京"
      }
    ],
    "Code": "100000",
    "Name": "中國"
  }
]

第三步：使用 AsTreeCte 遞歸查詢

若不做數據冗餘的無限級分類表設計，遞歸查詢少不了，AsTreeCte 正是解決遞歸查詢的封裝，方法參數說明：

參數 描述
(可選) pathSelector 路徑內容選擇，可以設置查詢返回：中國 -> 北京 -> 東城區
(可選) up false(默認)：由父級向子級的遞歸查詢，true：由子級向父級的遞歸查詢
(可選) pathSeparator 設置 pathSelector 的連接符，默認：->
(可選) level 設置遞歸層級

通過測試的數據庫：MySql8.0、SqlServer、PostgreSQL、Oracle、Sqlite、達夢、人大金倉

姿勢一：AsTreeCte() + ToTreeList

var t2 = fsql.Select<Area>()
  .Where(a => a.Name == "中國")
  .AsTreeCte() //查詢 中國 下的所有記錄
  .OrderBy(a => a.Code)
  .ToTreeList(); //非必須，也可以使用 ToList（見姿勢二）
Assert.Single(t2);
Assert.Equal("100000", t2[0].Code);
Assert.Single(t2[0].Childs);
Assert.Equal("110000", t2[0].Childs[0].Code);
Assert.Equal(2, t2[0].Childs[0].Childs.Count);
Assert.Equal("110100", t2[0].Childs[0].Childs[0].Code);
Assert.Equal("110101", t2[0].Childs[0].Childs[1].Code);
// WITH "as_tree_cte"
// as
// (
// SELECT 0 as cte_level, a."Code", a."Name", a."ParentCode" 
// FROM "Area" a 
// WHERE (a."Name" = '中國')

// union all

// SELECT wct1.cte_level + 1 as cte_level, wct2."Code", wct2."Name", wct2."ParentCode" 
// FROM "as_tree_cte" wct1 
// INNER JOIN "Area" wct2 ON wct2."ParentCode" = wct1."Code"
// )
// SELECT a."Code", a."Name", a."ParentCode" 
// FROM "as_tree_cte" a 
// ORDER BY a."Code"

姿勢二：AsTreeCte() + ToList

var t3 = fsql.Select<Area>()
  .Where(a => a.Name == "中國")
  .AsTreeCte()
  .OrderBy(a => a.Code)
  .ToList();
Assert.Equal(4, t3.Count);
Assert.Equal("100000", t3[0].Code);
Assert.Equal("110000", t3[1].Code);
Assert.Equal("110100", t3[2].Code);
Assert.Equal("110101", t3[3].Code);
//執行的 SQL 與姿勢一相同

姿勢三：AsTreeCte(pathSelector) + ToList

設置 pathSelector 參數后，如何返回隱藏字段？

var t4 = fsql.Select<Area>()
  .Where(a => a.Name == "中國")
  .AsTreeCte(a => a.Name + "[" + a.Code + "]")
  .OrderBy(a => a.Code)
  .ToList(a => new { 
    item = a, 
    level = Convert.ToInt32("a.cte_level"), 
    path = "a.cte_path" 
  });
Assert.Equal(4, t4.Count);
Assert.Equal("100000", t4[0].item.Code);
Assert.Equal("110000", t4[1].item.Code);
Assert.Equal("110100", t4[2].item.Code);
Assert.Equal("110101", t4[3].item.Code);
Assert.Equal("中國[100000]", t4[0].path);
Assert.Equal("中國[100000] -> 北京[110000]", t4[1].path);
Assert.Equal("中國[100000] -> 北京[110000] -> 北京市[110100]", t4[2].path);
Assert.Equal("中國[100000] -> 北京[110000] -> 東城區[110101]", t4[3].path);
// WITH "as_tree_cte"
// as
// (
// SELECT 0 as cte_level, a."Name" || '[' || a."Code" || ']' as cte_path, a."Code", a."Name", a."ParentCode" 
// FROM "Area" a 
// WHERE (a."Name" = '中國')

// union all

// SELECT wct1.cte_level + 1 as cte_level, wct1.cte_path || ' -> ' || wct2."Name" || '[' || wct2."Code" || ']' as cte_path, wct2."Code", wct2."Name", wct2."ParentCode" 
// FROM "as_tree_cte" wct1 
// INNER JOIN "Area" wct2 ON wct2."ParentCode" = wct1."Code"
// )
// SELECT a."Code" as1, a."Name" as2, a."ParentCode" as5, a.cte_level as6, a.cte_path as7 
// FROM "as_tree_cte" a 
// ORDER BY a."Code"

更多姿勢…請根據代碼註釋進行嘗試

源碼地址：https://github.com/dotnetcore/FreeSql

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※為什麼 USB CONNECTOR 是電子產業重要的元件?

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※新北清潔公司,居家、辦公、裝潢細清專業服務

※推薦評價好的iphone維修中心

前端開發神器Charles從入門到卸載

admin

前言

本文將帶大家學習使用前端開發神器-charles，從基本的下載安裝到常見配置使用，為大家一一講解。

一、花式誇獎Charles

  • 截取 Http 和 Https 網絡封包。
  • 支持重髮網絡請求，方便後端調試。
  • 支持修改網絡請求參數。
  • 支持網絡請求的截獲並動態修改。
  • 支持模擬慢速網絡。

好，騎上我心愛的小摩托，準備上路…

二、下載與安裝

官網下載傳送門

本文所使用的的版本為 mac V4.5.6版本，不同版本間的具體化差異，大家可留言交流。

Charles破解工具可通過關注公眾號「胡哥有話說」，回復關鍵字charles獲得。

三、簡單入門-抓包所有請求

  1. 打開Charles，勾選Proxy下的macOS Proxy（如果是windows，此處為Windows Proxy）
  2. 點擊Proxy->Start Recording，打開瀏覽器訪問任意頁面，可以在Charles中看到請求了。

很好，現在已經上路了，學習的路上永不堵車…

四、設置過濾請求

通過上面的操作，我們已經抓包了所有的請求，實際開發中可能是專門針對某些接口（如百度域名下的接口），我們可以專門配置過濾接口。

  1. 臨時性過濾配置

    在展示界面的Filter中可進行條件過濾

    同時可在右側的settings中配置使用正則來進行過濾

  2. 永久性過濾配置

    通過Proxy->Recording Settings->include中配置過濾條件

Ok，我們又前進了一大步

五、代理轉發請求

通過Charles的Map Remote和Map Local我們也可以配置代理轉發請求。

Map Remote

Map Remote 遠程映射，是將指定的網絡請求重定向到另一個網址

業務場景：
某些服務端的文件請求時限制某些特定域名(*.baidu.com)，我們使用localhost啟動項目時,會導致沒有權限訪問。通過配置Map Remote遠程映射解決問題。

配置路徑：
設置Tools->Map Remote

如圖上的配置，本地啟動的項目地址為：http://localhost:8080（或 http://127.0.0.1:8080），現在再訪問，可以使用路徑 http://test.baidu.com訪問即可。

注意Enable Map Remote一定要勾選，以及相應規則也要勾選，否則不會生效

Map Local

Map Local 本地映射，是指將指定的網絡請求重定向到本地的文件

業務場景：
在本地化的開發中，接口數據Mock；或者是線上環境問題排查時，將請求重定向到本地文件以方便排查。

配置路徑：
設置Tools->Map Local

通過如上圖的配置，請求 aa.baidu.com:443/index時，會被映射到本地 /xx/index.json

注意Enable Map Local一定要勾選，以及相關規則也要勾選，否則不會生效。

六、手機抓包

手機抓包請求也是我們日常開發中需要用到的，那如何利用Charles抓包手機請求呢。

  1. 設置Charles的代理端口號
    通過設置Proxy->Proxy Settings->Proxies->HTTP Proxy下的Port端口號

  2. 查看本地IP地址
    通過Charles的Help->Local IP Address查看，本機IP為xx.xx.xx.xx

  3. 手機和電腦需要處於同於wifi網絡內

  4. 手機wifi網絡配置

    以華為mate 30為例，選擇對應的wifi，選擇显示高級選項，設置代理為手動。
    設置服務器主機名為：xx.xx.xx.xx（剛才查看的電腦IP）
    設置服務器端口為：8888（剛才設置的port）
    點擊保存后，手機的請求就可以在Charles中查看啦…

    注意鏈接時，Charles會彈出授權窗口，要選擇Allow

七、限速設置

通過設置Proxy->Throttle Settings來進行速度限制

注意：一定要勾選 Enable Throttling選項

小結

以上是給大家分享的Charles的常見使用配置，如有相關問題可留言交流。

後記

以上就是胡哥今天給大家分享的內容，喜歡的小夥伴記得點贊、收藏呦，關注胡哥有話說，學習前端不迷路，歡迎多多留言交流…

胡哥有話說，專註於大前端技術領域，分享前端系統架構，框架實現原理，最新最高效的技術實踐！

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

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※幫你省時又省力,新北清潔一流服務好口碑

※回頭車貨運收費標準

ABP(ASP.NET Boilerplate Project)快速入門

admin

前言

這两天看了一下ABP，做個簡單的學習記錄。記錄主要有以下內容：

  1. 從官網創建並下載項目(.net core 3.x + vue)
  2. 項目在本地成功運行
  3. 新增實體並映射到數據庫
  4. 完成對新增實體的基本增刪改查

ABP官網：https://aspnetboilerplate.com/
Github：https://github.com/aspnetboilerplate

創建項目

進入官網

Get started，選擇前後端技術棧，我這裏就選.net core 3.x和vue。

填寫自己的項目名稱，郵箱，然後點create my project就可以下載項目了。

解壓文件

運行項目

後端項目

首先運行後端項目，打開/aspnet-core/MyProject.sln

改一下MyProject.Web.Host項目下appsettings.json的數據庫連接字符串，如果本地安裝了mssql，用windows身份認證，不改也行

數據庫默認是使用mssql的，當然也可以改其他數據庫。

將MyProject.Web.Host項目設置為啟動項，打開程序包管理器控制台，默認項目選擇DbContext所在的項目，也就是MyProject.EntityFrameworkCore。執行update-database

數據庫已成功創建：

Ctrl+F5，不出意外，瀏覽器就會看到這個界面：

前端項目

後端項目成功運行了，下面運行一下前端項目，先要確保本機有nodejs環境並安裝了vue cli，這個就不介紹了。

/vue目錄下打開cmd執行：npm install

install完成后執行：npm run serve

打開瀏覽器訪問http://localhost:8080/，不出意外的話，會看到這個界面：

使用默認用戶 admin/123qwe 登錄系統：

至此，前後端項目都已成功運行。
那麼基於abp的二次開發該從何下手呢，最簡單的，比如要增加一個數據表，並且完成最基本CRUD該怎麼做？

新增實體

實體類需要放在MyProject.Core項目下，我新建一個MyTest文件夾，並新增一個Simple類，隨意給2個屬性。

我這裏繼承了abp的Entity 類，Entity類有主鍵ID屬性，這個泛型int是指主鍵的類型，不寫默認就是int。abp還有一個比較複雜的FullAuditedEntity類型，繼承FullAuditedEntity的話就有創建時間，修改時間，創建人，修改人，軟刪除等字段。這個看實際情況。 

public class Simple : Entity<int>
{
    public string Name { get; set; }

    public string Details { get; set; }
}

修改MyProject.EntityFrameworkCore項目的/EntityFrameworkCore/MyProjectDbContext：

public class MyProjectDbContext : AbpZeroDbContext<Tenant, Role, User, MyProjectDbContext>
{
    /* Define a DbSet for each entity of the application */

    public DbSet<Simple> Simples { get; set; }

    public MyProjectDbContext(DbContextOptions<MyProjectDbContext> options)
        : base(options)
    {
    }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);

        modelBuilder.Entity<Simple>(p =>
        {
            p.ToTable("Simples", "test");
            p.Property(x => x.Name).IsRequired(true).HasMaxLength(20);
            p.Property(x => x.Details).HasMaxLength(100);
        });
    }
}

然後就可以遷移數據庫了，程序包管理器控制台執行：add-migration mytest1，update-database

刷新數據庫，Simples表已生成：

實體的增刪改查

進入MyProject.Application項目，新建一個MyTest文件夾

Dto

CreateSimpleDto，新增Simple數據的傳輸對象，比如ID，創建時間，創建人等字段，就可以省略

public class CreateSimpleDto
{
    public string Name { get; set; }

    public string Details { get; set; }
}

PagedSimpleResultRequestDto，分頁查詢對象

public class PagedSimpleResultRequestDto : PagedResultRequestDto
{
    /// <summary>
    /// 查詢關鍵字
    /// </summary>
    public string Keyword { get; set; }
}

SimpleDto，這裏跟CreateSimpleDto的區別就是繼承了EntityDto，多了個ID屬性

public class SimpleDto : EntityDto<int>
{
    public string Name { get; set; }

    public string Details { get; set; }
}

SimpleProfile，用來定義AutoMapper的映射關係清單

public class SimpleProfile : Profile
{
    public SimpleProfile()
    {
        CreateMap<Simple, SimpleDto>();
        CreateMap<SimpleDto, Simple>();
        CreateMap<CreateSimpleDto, Simple>();
    }
}

Service

注意，類名參考abp的規範去命名。

ISimpleAppService，Simple服務接口。我這裏繼承IAsyncCrudAppService，這個接口中包含了增刪改查的基本定義，非常方便。如果不需要的話，也可以繼承IApplicationService自己定義

public interface ISimpleAppService : IAsyncCrudAppService<SimpleDto, int, PagedSimpleResultRequestDto, CreateSimpleDto, SimpleDto>
{

}

SimpleAppService，Simple服務，繼承包含了增刪改查的AsyncCrudAppService類，如果有需要的話可以override這些增刪改查方法。也可以繼承MyProjectAppServiceBase，自己定義。

public class SimpleAppService : AsyncCrudAppService<Simple, SimpleDto, int, PagedSimpleResultRequestDto, CreateSimpleDto, SimpleDto>, ISimpleAppService
{
    public SimpleAppService(IRepository<Simple, int> repository) : base(repository)
    {

    }

    /// <summary>
    /// 條件過濾
    /// </summary>
    /// <param name="input"></param>
    /// <returns></returns>
    protected override IQueryable<Simple> CreateFilteredQuery(PagedSimpleResultRequestDto input)
    {
        return Repository.GetAll()
            .WhereIf(!input.Keyword.IsNullOrWhiteSpace(), a => a.Name.Contains(input.Keyword));
    }
}

接口測試

重新運行項目，不出意外的話，Swagger中就會多出Simple相關的接口。

  • Create

  • Get

  • GetAll

  • Update

  • Delete

總結

ABP是一個優秀的框架，基於ABP的二次開發肯定會非常高效，但前提是需要熟練掌握ABP，弄清楚他的設計理念以及他的一些實現原理。

以後有時間的話再深入學習一下。文中如果有不妥之處歡迎指正。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※南投搬家公司費用需注意的眉眉角角，別等搬了再說!

※新北清潔公司,居家、辦公、裝潢細清專業服務

認證授權方案之授權初識

admin

1.前言

回顧：認證授權方案之JwtBearer認證

在上一篇中，我們通過JwtBearer的方式認證，了解在認證時，都是基於Claim的，因此我們可以通過用戶令牌獲取到用戶的Claims，在授權過程中對這些Claims進行驗證，從而來判斷是否具有獲取或執行目標資源操作的權限。本章就來介紹一下 ASP.NET Core 的授權系統的簡單使用。

2.說明

授權與身份認證是相互獨立，但是，授權卻需要一種身份驗證機制，因此，身份驗證可以為當前用戶創建一個或多個標識，是確定用戶真實身份的過程。而授權是根據標識確定用戶可執行的操作的過程，其本質就是具有某種特性的用戶會有權限訪問某個資源或者執行某個操作。例如：一個擁有管理員身份的用戶有創建人員、刪除人員、編輯人員和刪除人員的操作權限，而一個非管理身份的用戶僅有讀取自己信息的權限。

這時候，你可能會問，究竟怎樣特性的用戶可以被授權訪問某個資源或執行某個操作。由此我們引出了授權策略的方式，可以根據用戶擁有的角色，也可以根據用戶的職位，部門甚至是性別，年齡等等特性進行授權。

通過建立授權策略方式，檢驗認證的用戶所攜帶的身份聲明（ClaimsPrincipal對象）與授權策略是否一致，從而確定用戶可否執行操作。

3.授權

3.1. 基於角色

3.1.1 添加角色

將角色賦予某個控制器或控制器內的操作，指定當前用戶必須是其角色才能訪問請求資源。

可以使用Authorize屬性的Roles特性指定所請求資源的角色。

例如：

  • 分配了“admin”角色用戶進行訪問操作
[Authorize(Roles ="admin")]
public class WeatherForecastController : ControllerBase
{

}
  • 以逗號分隔角色名來允行多個角色訪問操作
[Authorize(Roles ="admin,user")]
public class WeatherForecastController : ControllerBase
{ 


}

其中只要滿足admmin或者user其一就可以進行訪問。

  • 同時滿足指定的多個角色進行的訪問操作
[Authorize(Roles = "admin")]
[Authorize(Roles = "user")]
public class WeatherForecastController : ControllerBase
{ 
}

3.1.2 添加策略的角色

可以創建策略的方式進行訪問控制，在配置授權服務中添加註冊授權服務策略。

在Startup.cs文件中，通過ConfigureServices()配置服務,創建一個允許具有admin角色的用戶才能進行訪問的策略

    public void ConfigureServices(IServiceCollection services)
    {
        services.AddControllers();
        //添加授權角色策略
        services.AddAuthorization(options =>
        {
            options.AddPolicy("BaseRole", options => options.RequireRole("admin"));
        });
        //或者指定多個允許的角色
        //services.AddAuthorization(options =>
        // {
        //    options.AddPolicy("MoreBaseRole", options => options.RequireRole("admin","user"));
        // });
    }

在控制器方法使用特性Policy的屬性進行策略應用

    [Authorize(Policy = "BaseRole")]
    public class WeatherForecastController : ControllerBase
    {
    
    }

3.2. 基於聲明

3.2.1添加聲明

對當前用戶必須擁有的聲明，並將聲明賦予某個控制器或控制器內的操作，因此，指定聲明必須持有對應的值才能訪問請求資源。

聲明要求基於策略，所以必須進行構建一個表示聲明要求的策略，才能進行授權。

最簡單的類型聲明是將判斷聲明是否存在，而不檢查值。

可以創建策略的方式進行訪問控制，在配置授權服務中添加註冊授權服務策略。

在Startup.cs文件中，通過ConfigureServices()配置服務,創建一個允許具有聲明的用戶才能進行訪問的策略

    public void ConfigureServices(IServiceCollection services)
    {
        services.AddControllers();
        //添加基於聲明的授權
        services.AddAuthorization(options =>
        {
            options.AddPolicy("BaseClaims", options => options.RequireClaim("name"));
        });
    }

BaseClaims聲明策略會檢查name當前標識是否存在聲明。

在控制器方法使用特性Policy的屬性進行策略應用

    [Authorize(Policy = "BaseClaims")]
    public class WeatherForecastController : ControllerBase
    {
    
    }

但是，大多時候，我們需要聲明包含值，只有指定允許值的列表，才能授權成功。所以，可以添加指定值。

    public void ConfigureServices(IServiceCollection services)
    {
        services.AddControllers();
        //添加基於聲明的授權,指定允許值列表。
        services.AddAuthorization(options =>
        {
            options.AddPolicy("BaseClaims", options => options.RequireClaim("name","i3yuan"));
        });
    }

3.3 基於策略

上面介紹的基於角色和基於聲明的授權，都使用了要求、要求處理程序和預配置的策略。這些在構建上提供了便捷，但是最終都是生成授權策略。ASP.NET Core，設計了另一種靈活的授權方式，一種更豐富的可重複使用的授權結構，基於策略的授權，同時這也是授權的核心。

這節會先講一下授權策略的應用，在下一節中，會對授權策略的核心進行一步步的詳解。

在上面我們簡單的介紹了基於策略的角色授權，但是這種方式無非基於角色或者聲明多一些。

因此，這裏我們基於自定義策略授權的方式，實現授權。

自定義授權，就要我們自己寫策略提供器，自己根據不同的參數來生成不同的策略，重新實現策略的方式。策略要求由以下兩種元素組成：僅保留數據的要求類，以及對用戶驗證數據的授權處理程序。創建自定義要求，還可以進一步表達特定策略。

3.3.1. 定義權限策略PermissionRequirement

定義一個權限策略，這個策略並包含一些屬性。

public class PermissionRequirement: IAuthorizationRequirement
{
    public string _permissionName { get; }

    public PermissionRequirement(string PermissionName)
    {
        _permissionName = PermissionName;
    }
}

3.3.2. 再定義一個策略處理類

public class PermissionRequirementHandler : AuthorizationHandler<PermissionRequirement>
{
    protected override Task HandleRequirementAsync(AuthorizationHandlerContext context, PermissionRequirement requirement)
    {
        var role = context.User.FindFirst(c => c.Type == ClaimTypes.Role);
        if (role != null)
        {
            var roleValue = role.Value;
            if (roleValue==requirement._permissionName)
            {
                context.Succeed(requirement);
            }
        }
        return Task.CompletedTask;

授權處理程序讀取與角色用戶關聯的聲明，並檢查自定義的角色，如果角色匹則成功，否則無法返回成功。

這裏的自定義聲明是寫固定了，但是也可以通過數據庫或外部服務的方式進行運行查詢獲取用戶相關角色信息相對應的判斷條件，從而在處理程序中進行判斷處理。

授權處理程序調用方法 Succeed，同時傳遞當前要求，以通知此要求已成功得到驗證。如果沒有傳遞要求，處理程序無需執行任何操作，可以直接返回內容。不過，如果處理程序要確定是否不符合要求（無論其他處理程序是否已成功驗證同一要求），將會對授權上下文對象調用方法 Fail。

3.3.3. 下面展示了如何將自定義要求添加到策略

（請注意，由於這是自定義要求，因此沒有擴展方法，而必須繼續處理策略對象的整個 Requirements 集合）：

    public void ConfigureServices(IServiceCollection services)
    {
        services.AddControllers();
        //基於自定義策略授權
        services.AddAuthorization(options =>
        {
            options.AddPolicy("customizePermisson",
              policy => policy
                .Requirements
                .Add(new PermissionRequirement("admin")));
        });
        //此外，還需要在 IAuthorizationHandler 類型的範圍內向 DI 系統註冊新的處理程序：
        services.AddScoped<IAuthorizationHandler, PermissionRequirementHandler>();
        // 如前所述，要求可包含多個處理程序。如果為授權層的同一要求向 DI 系統註冊多個處理程序，有一個成功就足夠了。

    }

3.3.4. 應用自定義的策略的特性

指定當前用戶必須是應用對控制器或控制器內的操作，如

   [Authorize(Policy = "customizePermisson")]
    public class WeatherForecastController : ControllerBase
    { 
    }

4.場景

在上一篇認證授權方案之JwtBearer認證中，我們已經實現了獲取token的方式，這一次，我們實現一個以基於角色場景為例的認證授權。

在原來生成token的方式中，添加多一個聲明角色的Claim，如下：

new Claim(JwtClaimTypes.Role,”admin”)

    [HttpGet]
    public IActionResult GetToken()
    {
        try
        {
            //定義發行人issuer
            string iss = "JWTBearer.Auth";
            //定義受眾人audience
            string aud = "api.auth";
            //定義許多種的聲明Claim,信息存儲部分,Claims的實體一般包含用戶和一些元數據
            IEnumerable<Claim> claims = new Claim[]
            {
                new Claim(JwtClaimTypes.Id,"1"),
                new Claim(JwtClaimTypes.Name,"i3yuan"),
                new Claim(JwtClaimTypes.Role,"admin"),
            };
            //notBefore  生效時間
            // long nbf =new DateTimeOffset(DateTime.Now).ToUnixTimeSeconds();
            var nbf = DateTime.UtcNow;
            //expires   //過期時間
            // long Exp = new DateTimeOffset(DateTime.Now.AddSeconds(1000)).ToUnixTimeSeconds();
            var Exp = DateTime.UtcNow.AddSeconds(1000);
            //signingCredentials  簽名憑證
            string sign = "q2xiARx$4x3TKqBJ"; //SecurityKey 的長度必須 大於等於 16個字符
            var secret = Encoding.UTF8.GetBytes(sign);
            var key = new SymmetricSecurityKey(secret);
            var signcreds = new SigningCredentials(key, SecurityAlgorithms.HmacSha256);
            //String issuer = default(String), String audience = default(String), IEnumerable<Claim> claims = null, Nullable<DateTime> notBefore = default(Nullable<DateTime>), Nullable<DateTime> expires = default(Nullable<DateTime>), SigningCredentials signingCredentials = null
            var jwt = new JwtSecurityToken(issuer: iss, audience: aud, claims:claims,notBefore:nbf,expires:Exp, signingCredentials: signcreds);
            var JwtHander = new JwtSecurityTokenHandler();
            var token = JwtHander.WriteToken(jwt);
            return Ok(new
            {
                access_token = token,
                token_type = "Bearer",
            });
        }
        catch (Exception ex)
        {
            throw;
        }
    }

對控制器或控制器內的操作，指定當前用戶必須是其角色才能訪問請求資源，如WeatherForecastController.cs

[ApiController]
[Route("[controller]")]
[Authorize(Roles ="admin")]
public class WeatherForecastController : ControllerBase
{ 
    private static readonly string[] Summaries = new[]
    {
        "Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Sweltering", "Scorching"
    };

    private readonly ILogger<WeatherForecastController> _logger;

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

    [HttpGet]
    public IEnumerable<WeatherForecast> Get()
    {
        var rng = new Random();
        return Enumerable.Range(1, 5).Select(index => new WeatherForecast
        {
            Date = DateTime.Now.AddDays(index),
            TemperatureC = rng.Next(-20, 55),
            Summary = Summaries[rng.Next(Summaries.Length)]
        })
        .ToArray();
    }
}

5.運行

5.1. 獲取token

分別獲取role為admin和role為user的情況下頒發的token，只有在角色為admin的情況下才能授權通過。

5.2. 授權資源接口訪問

在role為admin的情況下

在role為user的情況下

由上可知，只有在角色為admin的情況下，才能訪問目標資源進行操作。

6.總結

  1. 從上一篇的認證到這一篇的授權階段，簡單的介紹了Asp.net Core的認證授權系統，對授權有了初步的認識以及使用，對授權進行劃分為兩種，一種是基於角色的授權，但隨着角色的增加會對處理授權產生限制，不適合表達複雜的授權邏輯。另一種是基於策略的身份驗證，策略包含一系列基於聲明的要求，以及基於可從 HTTP 上下文或外部源注入的其他任何信息的自定義邏輯。這些要求各自與一個或多個處理程序相關聯，這些處理程序負責要求的實際計算。
  2. 可以發現，asp.net core提供的授權策略是一個非常強大豐富且靈活的認證授權方案，能夠滿足大部分的授權場景。
  3. 如果有不對的或不理解的地方，希望大家可以多多指正，提出問題，一起討論,不斷學習,共同進步。
  4. 因此，在後續的篇章中，會繼續探索授權系統，對授權策略的核心進行一步步的詳解。
  5. 本示例源碼地址

參考文獻文檔

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※想知道最厲害的網頁設計公司"嚨底家"!

※幫你省時又省力,新北清潔一流服務好口碑

※別再煩惱如何寫文案,掌握八大原則!

不再逼動物賺錢 吳哥窟園區2020年起禁止騎象

admin

摘錄自2019年11月15日中央通訊社吳哥窟報導

柬埔寨的吳哥窟暨暹粒地區保護管理局（Apsara Authority）官員今天（15日）表示，明年初開始將禁止吳哥窟所在考古園區的大象騎乘活動。

吳哥考古園區位在柬埔寨北部，匯集柬國大多數外籍遊客，去年創下約600萬造訪人次新高紀錄。不少遊客選擇騎乘大象在園區內的古老寺廟遺址間漫遊。吳哥窟暨暹粒地區保護管理局發言人告訴法新社：「藉大象營利已不再恰當。」他還說，部分大象年事已高。

園區原本有14頭「在職」大象，其中5頭已被送到距離寺廟古蹟群大約40公里的樹林安置。管理局發言人指出：「牠們在那裡將以自然方式度過餘生。」他強調，擁有這些大象的企業將持續照護牠們。

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※為什麼 USB CONNECTOR 是電子產業重要的元件?

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※新北清潔公司,居家、辦公、裝潢細清專業服務

※推薦評價好的iphone維修中心

佛羅里達州九成柑橘染「愛滋」 三分之二果汁停產

admin

摘錄自2019年11月14日法廣社報導

美國的重要柑橘產區佛羅里達州九成柑橘樹感染源自中國的致命病害「黃龍病」，導致樹枝枯萎、葉片扭曲變形、樹幹腐爛；專家將這種植物病害稱之為「柑橘愛滋病」。

佛羅里達州柑橘產業價值90億元，為該州第二大產業，僅次於觀光業；佛州的柳橙汁產量佔全國逾80%。《華盛頓郵報》報導，佛州當局稱黃龍病為「可預期破壞性最大的外來植物病害之一」，已摧毀佛州標誌性的柑橘產業。

隨着每年11月到隔年5月的採收季節到來，「幽靈樹叢」卻遍布佛州各地，迫使農民廢棄部分作物，結束果汁生意，甚至將整座事業轉賣；《華盛頓郵報》報導說，2004年佛州有7000多名農民種植柑橘，如今約5000人已經放棄。

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※幫你省時又省力,新北清潔一流服務好口碑

※回頭車貨運收費標準

世界最髒河水可飲用 印尼拚2025乾淨家園

admin

摘錄自2019年11月18日中央通訊社報導

西爪哇的西大魯河曾被稱為世界上最髒的河流，印尼從2年前整治至今，部分上游河水已可供居民飲用。印尼官員說，乾淨印尼是全國性運動，將持續動員全民達到2025年的目標。

印尼總統佐科威（Joko Widodo）在2017年宣示，2025年將達成讓印尼乾淨的目標（Gerakan Indonesia Bersih），包括垃圾從源頭減量3成、並處理7成的垃圾，以避免垃圾直接放置於堆積場或流入海洋。

這是印尼首度、也是唯一動員軍隊清理的河川。負責清理的少將蘇山托（Susanto）之前表示，上游整治後，經養魚測試水質6個月前設置第一台淨水機器，已開始供居民使用。

印尼人口2億6700萬，因垃圾清運不普及、缺乏處理機制，也沒有回收概念，大量垃圾不是丟河裡，就是掩埋或燒掉。許多都市貧民社區及鄉下民眾取用浸泡垃圾的河水作日常使用，暴露在遭垃圾污染的環境及空氣中，衛生條件極差。

輿論普遍分析，印尼乾淨運動是遠大、有難度的目標。印尼海洋事務統籌部海洋科學與科技主任納尼說，政府已列出67條優先治理的河川，以減少海洋垃圾；密集與地方政府合作推動減量，目前已有峇里島等12個地方政府制定限制使用一次性塑膠產品；也研擬嚴懲與觀光有關的行為造成的垃圾污染；並尋求新科技的運用。

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

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

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※南投搬家公司費用需注意的眉眉角角，別等搬了再說!

※新北清潔公司,居家、辦公、裝潢細清專業服務

減塑全球風潮 2020年起泰國商場禁用塑膠袋

admin

摘錄自2019年11月18日NOWnews報導

為了達到環保的目的，2020年起泰國許多連鎖超商或大賣場將不再供應塑膠袋。根據路透社的報導，泰國國家資源和環境部部長瓦拉伍特（Varawut Silpa-archa）在臉書宣佈，許多商店與政府合作，暫停提供一次用塑膠袋，措施將於2020年1月起生效。

「方便性會下降，不過這是為了延續環境的生態循環。」瓦拉伍特受訪時說，他建議民眾可以使用可重複利用的大布袋取代塑膠袋。

不再提供塑膠袋的商店包括在泰國管理上千家7-11的CP All Plc公司和泰國購物商場集團。其實，這並不是泰國第一次嘗試縮減塑膠袋使用的政策。早在2018年，CAN便報導過泰國旅遊局曾發佈「國家淨化聲明」，希望在2021年時，能減少一半以上現行使用的塑膠袋總用量。

本站聲明:網站內容來源環境資訊中心https://e-info.org.tw/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※想知道最厲害的網頁設計公司"嚨底家"!

※幫你省時又省力,新北清潔一流服務好口碑

※別再煩惱如何寫文案,掌握八大原則!

Spring Boot 2.x基礎教程：MyBatis的多數據源配置

admin

前两天，我們已經介紹了關於JdbcTemplate的多數據源配置以及Spring Data JPA的多數據源配置，接下來具體說說使用MyBatis時候的多數據源場景該如何配置。

添加多數據源的配置

先在Spring Boot的配置文件application.properties中設置兩個你要鏈接的數據庫配置，比如這樣：

spring.datasource.primary.jdbc-url=jdbc:mysql://localhost:3306/test1
spring.datasource.primary.username=root
spring.datasource.primary.password=123456
spring.datasource.primary.driver-class-name=com.mysql.cj.jdbc.Driver

spring.datasource.secondary.jdbc-url=jdbc:mysql://localhost:3306/test2
spring.datasource.secondary.username=root
spring.datasource.secondary.password=123456
spring.datasource.secondary.driver-class-name=com.mysql.cj.jdbc.Driver

說明與注意：

  1. 多數據源配置的時候，與單數據源不同點在於spring.datasource之後多設置一個數據源名稱primary和secondary來區分不同的數據源配置，這個前綴將在後續初始化數據源的時候用到。
  2. 數據源連接配置2.x和1.x的配置項是有區別的：2.x使用spring.datasource.secondary.jdbc-url，而1.x版本使用spring.datasource.secondary.url。如果你在配置的時候發生了這個報錯java.lang.IllegalArgumentException: jdbcUrl is required with driverClassName.，那麼就是這個配置項的問題。
  3. 可以看到，不論使用哪一種數據訪問框架，對於數據源的配置都是一樣的。

初始化數據源與MyBatis配置

完成多數據源的配置信息之後，就來創建個配置類來加載這些配置信息，初始化數據源，以及初始化每個數據源要用的MyBatis配置。

這裏我們繼續將數據源與框架配置做拆分處理：

  1. 單獨建一個多數據源的配置類，比如下面這樣：
@Configuration
public class DataSourceConfiguration {

    @Primary
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.primary")
    public DataSource primaryDataSource() {
        return DataSourceBuilder.create().build();
    }

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.secondary")
    public DataSource secondaryDataSource() {
        return DataSourceBuilder.create().build();
    }

}

可以看到內容跟JdbcTemplate、Spring Data JPA的時候是一模一樣的。通過@ConfigurationProperties可以知道這兩個數據源分別加載了spring.datasource.primary.*和spring.datasource.secondary.*的配置。@Primary註解指定了主數據源，就是當我們不特別指定哪個數據源的時候，就會使用這個Bean真正差異部分在下面的JPA配置上。

  1. 分別創建兩個數據源的MyBatis配置。

Primary數據源的JPA配置：

@Configuration
@MapperScan(
        basePackages = "com.didispace.chapter39.p",
        sqlSessionFactoryRef = "sqlSessionFactoryPrimary",
        sqlSessionTemplateRef = "sqlSessionTemplatePrimary")
public class PrimaryConfig {

    private DataSource primaryDataSource;

    public PrimaryConfig(@Qualifier("primaryDataSource") DataSource primaryDataSource) {
        this.primaryDataSource = primaryDataSource;
    }

    @Bean
    public SqlSessionFactory sqlSessionFactoryPrimary() throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(primaryDataSource);
        return bean.getObject();
    }

    @Bean
    public SqlSessionTemplate sqlSessionTemplatePrimary() throws Exception {
        return new SqlSessionTemplate(sqlSessionFactoryPrimary());
    }

}

Secondary數據源的JPA配置：

@Configuration
@MapperScan(
        basePackages = "com.didispace.chapter39.s",
        sqlSessionFactoryRef = "sqlSessionFactorySecondary",
        sqlSessionTemplateRef = "sqlSessionTemplateSecondary")
public class SecondaryConfig {

    private DataSource secondaryDataSource;

    public SecondaryConfig(@Qualifier("secondaryDataSource") DataSource secondaryDataSource) {
        this.secondaryDataSource = secondaryDataSource;
    }

    @Bean
    public SqlSessionFactory sqlSessionFactorySecondary() throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(secondaryDataSource);
        return bean.getObject();
    }

    @Bean
    public SqlSessionTemplate sqlSessionTemplateSecondary() throws Exception {
        return new SqlSessionTemplate(sqlSessionFactorySecondary());
    }

}

說明與注意：

  1. 配置類上使用@MapperScan註解來指定當前數據源下定義的Entity和Mapper的包路徑；另外需要指定sqlSessionFactory和sqlSessionTemplate，這兩個具體實現在該配置類中類中初始化。
  2. 配置類的構造函數中，通過@Qualifier註解來指定具體要用哪個數據源，其名字對應在DataSourceConfiguration配置類中的數據源定義的函數名。
  3. 配置類中定義SqlSessionFactory和SqlSessionTemplate的實現，注意具體使用的數據源正確（如果使用這裏的演示代碼，只要第二步沒問題就不需要修改）。

上一篇介紹JPA的時候，因為之前介紹JPA的使用時候，說過實體和Repository定義的方法，所以省略了 User 和 Repository的定義代碼，但是還是有讀者問怎麼沒有這個，其實都有說明，倉庫代碼里也都是有的。未避免再問這樣的問題，所以這裏就貼一下吧。

根據上面Primary數據源的定義，在com.didispace.chapter39.p包下，定義Primary數據源要用的實體和數據訪問對象，比如下面這樣：

@Data
@NoArgsConstructor
public class UserPrimary {

    private Long id;

    private String name;
    private Integer age;

    public UserPrimary(String name, Integer age) {
        this.name = name;
        this.age = age;
    }
}

public interface UserMapperPrimary {

    @Select("SELECT * FROM USER WHERE NAME = #{name}")
    UserPrimary findByName(@Param("name") String name);

    @Insert("INSERT INTO USER(NAME, AGE) VALUES(#{name}, #{age})")
    int insert(@Param("name") String name, @Param("age") Integer age);

    @Delete("DELETE FROM USER")
    int deleteAll();

}

根據上面Secondary數據源的定義，在com.didispace.chapter39.s包下，定義Secondary數據源要用的實體和數據訪問對象，比如下面這樣：

@Data
@NoArgsConstructor
public class UserSecondary {

    private Long id;

    private String name;
    private Integer age;

    public UserSecondary(String name, Integer age) {
        this.name = name;
        this.age = age;
    }
}

public interface UserMapperSecondary {

    @Select("SELECT * FROM USER WHERE NAME = #{name}")
    UserSecondary findByName(@Param("name") String name);

    @Insert("INSERT INTO USER(NAME, AGE) VALUES(#{name}, #{age})")
    int insert(@Param("name") String name, @Param("age") Integer age);

    @Delete("DELETE FROM USER")
    int deleteAll();
}

測試驗證

完成了上面之後，我們就可以寫個測試類來嘗試一下上面的多數據源配置是否正確了，先來設計一下驗證思路：

  1. 往Primary數據源插入一條數據
  2. 從Primary數據源查詢剛才插入的數據，配置正確就可以查詢到
  3. 從Secondary數據源查詢剛才插入的數據，配置正確應該是查詢不到的
  4. 往Secondary數據源插入一條數據
  5. 從Primary數據源查詢剛才插入的數據，配置正確應該是查詢不到的
  6. 從Secondary數據源查詢剛才插入的數據，配置正確就可以查詢到

具體實現如下：

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
@Transactional
public class Chapter39ApplicationTests {

    @Autowired
    private UserMapperPrimary userMapperPrimary;
    @Autowired
    private UserMapperSecondary userMapperSecondary;

    @Before
    public void setUp() {
        // 清空測試表，保證每次結果一樣
        userMapperPrimary.deleteAll();
        userMapperSecondary.deleteAll();
    }

    @Test
    public void test() throws Exception {
        // 往Primary數據源插入一條數據
        userMapperPrimary.insert("AAA", 20);

        // 從Primary數據源查詢剛才插入的數據，配置正確就可以查詢到
        UserPrimary userPrimary = userMapperPrimary.findByName("AAA");
        Assert.assertEquals(20, userPrimary.getAge().intValue());

        // 從Secondary數據源查詢剛才插入的數據，配置正確應該是查詢不到的
        UserSecondary userSecondary = userMapperSecondary.findByName("AAA");
        Assert.assertNull(userSecondary);

        // 往Secondary數據源插入一條數據
        userMapperSecondary.insert("BBB", 20);

        // 從Primary數據源查詢剛才插入的數據，配置正確應該是查詢不到的
        userPrimary = userMapperPrimary.findByName("BBB");
        Assert.assertNull(userPrimary);

        // 從Secondary數據源查詢剛才插入的數據，配置正確就可以查詢到
        userSecondary = userMapperSecondary.findByName("BBB");
        Assert.assertEquals(20, userSecondary.getAge().intValue());
    }

}

代碼示例

本文的相關例子可以查看下面倉庫中的chapter3-9目錄：

  • Github：https://github.com/dyc87112/SpringBoot-Learning/
  • Gitee：https://gitee.com/didispace/SpringBoot-Learning/

如果您覺得本文不錯，歡迎Star支持，您的關注是我堅持的動力！

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※為什麼 USB CONNECTOR 是電子產業重要的元件?

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※台北網頁設計公司全省服務真心推薦

※想知道最厲害的網頁設計公司"嚨底家"!

※新北清潔公司,居家、辦公、裝潢細清專業服務

※推薦評價好的iphone維修中心

RocketMQ入門到入土（一）新手也能看懂的原理和實戰！

admin

學任何技術都是兩步驟：

  1. 搭建環境

  2. helloworld

我也不例外，直接搞起來。

一、RocketMQ的安裝

1、文檔

官方網站

http://rocketmq.apache.org

GitHub

https://github.com/apache/rocketmq

2、下載

wget https://mirror.bit.edu.cn/apache/rocketmq/4.7.0/rocketmq-all-4.7.0-bin-release.zip

我們是基於Centos8來的，面向官方文檔學習，所以下載地址自然也是官方的。

去官方網站找合適的版本進行下載，目前我這裏最新的是4.7.0版本。

http://rocketmq.apache.org/dowloading/releases/

 

https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.7.0/rocketmq-all-4.7.0-bin-release.zip

 

3、準備工作

3.1、解壓

unzip rocketmq-all-4.7.0-bin-release.zip

3.2、安裝jdk

sudo yum install java-1.8.0-openjdk-devel

4、啟動

4.1、啟動namesrv

cd rocketmq-all-4.7.0-bin-release/bin
./mqnamesrv

4.2、啟動broker

cd rocketmq-all-4.7.0-bin-release/bin
./mqbroker -n localhost:9876

常見錯誤以及解決方案：

常見錯誤：啟動broker失敗 Cannot allocate memory

[root@node-113b bin]# ./mqbroker -n localhost:9876
Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x00000005c0000000, 8589934592, 0) failed
; error='Cannot allocate memory' (errno=12)#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /usr/local/rocketmq/bin/hs_err_pid1997.log

解決方案：

是由於默認內存分配的太大了，超出了本機內存，直接OOM了。

修改bin/目錄下的如下兩個腳本

runbroker.sh
runserver.sh

在這兩個腳本里都搜索-server -Xms，將其內存分配小點，自己玩的話512MB就足夠了，夠夠的了！

4.3、啟動成功標識

namesrv啟動成功標識：

broker啟動成功標識：

二、RocketMQ控制台的安裝

控制台目前獲取方式有如下兩種：

  1. 第三方網站去下載現成的，比如csdn等。

  2. 官方源碼包自己編譯而成，官方沒有現成的。

我們這裏當然採取官方方式。

1、官方文檔

github倉庫

https://github.com/apache/rocketmq-externals

中文指南

https://github.com/apache/rocketmq-externals/blob/master/rocketmq-console/doc/1_0_0/UserGuide_CN.md

2、下載源碼

https://codeload.github.com/apache/rocketmq-externals/zip/master

3、修改配置（可選）

我們下載完解壓后的文件目錄如下：

修改rocketmq-consolesrcmainresourcesapplication.properties文件的server.port就歐了。默認8080。

4、編譯打包

進入rocketmq-console，然後用maven進行編譯打包

mvn clean package -DskipTests

打包完會在target下生成我們spring boot的jar程序，直接java -jar啟動完事。

5、啟動控制台

將編譯打包好的springboot程序扔到服務器上，執行如下命令進行啟動

java -jar rocketmq-console-ng-1.0.1.jar --rocketmq.config.namesrvAddr=127.0.0.1:9876

如果想後台啟動就nohup &

訪問一下看看效果：

三、測試

rocketmq給我們提供了測試工具和測試類，可以在安裝完很方便的進行測試。

0、準備工作

rocketmq給我們提供的默認測試工具在bin目錄下，叫tools.sh。我們測試前需要配置這個腳本，為他指定namesrv地址才可以，否則測試發送/消費消息的時候會出現如下錯誤 connect to null failed：

22:49:02.470 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARN Please initialize the logger system properly.
java.lang.IllegalStateException: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to null failed

配置如下：

vim tools.sh
# 在export JAVA_HOME上面添加如下這段代碼
export NAMESRV_ADDR=localhost:9876

1、發送消息

./tools.sh org.apache.rocketmq.example.quickstart.Producer

成功的話會看到嘩嘩嘩的日誌，因為這個類會發送1000條消息到TopicTest這個Topic下。

2、消費消息

./tools.sh org.apache.rocketmq.example.quickstart.Consumer

成功的話會看到嘩嘩嘩的日誌，因為這個類會消費TopicTest下的全部消息。剛發送的1000條都會被消費掉。

3、控制台

發送成功后我們自然也能來到管控台去看消息和消費情況等等等信息

四、架構圖以及角色

1、架構圖

2、角色

2.1、Broker

  • 理解成RocketMQ本身

  • broker主要用於producer和consumer接收和發送消息

  • broker會定時向nameserver提交自己的信息

  • 是消息中間件的消息存儲、轉發服務器

  • 每個Broker節點，在啟動時，都會遍歷NameServer列表，與每個NameServer建立長連接，註冊自己的信息，之後定時上報

2.2、Nameserver

  • 理解成zookeeper的效果，只是他沒用zk，而是自己寫了個nameserver來替代zk

  • 底層由netty實現，提供了路由管理、服務註冊、服務發現的功能，是一個無狀態節點

  • nameserver是服務發現者，集群中各個角色（producer、broker、consumer等）都需要定時向nameserver上報自己的狀態，以便互相發現彼此，超時不上報的話，nameserver會把它從列表中剔除

  • nameserver可以部署多個，當多個nameserver存在的時候，其他角色同時向他們上報信息，以保證高可用，

  • NameServer集群間互不通信，沒有主備的概念

  • nameserver內存式存儲，nameserver中的broker、topic等信息默認不會持久化，所以他是無狀態節點

2.3、Producer

  • 消息的生產者

  • 隨機選擇其中一個NameServer節點建立長連接，獲得Topic路由信息（包括topic下的queue，這些queue分佈在哪些broker上等等）

  • 接下來向提供topic服務的master建立長連接（因為rocketmq只有master才能寫消息），且定時向master發送心跳

2.4、Consumer

  • 消息的消費者

  • 通過NameServer集群獲得Topic的路由信息，連接到對應的Broker上消費消息

  • 由於Master和Slave都可以讀取消息，因此Consumer會與Master和Slave都建立連接進行消費消息

3、核心流程

  • Broker都註冊到Nameserver上

  • Producer發消息的時候會從Nameserver上獲取發消息的topic信息

  • Producer向提供服務的所有master建立長連接，且定時向master發送心跳

  • Consumer通過NameServer集群獲得Topic的路由信息

  • Consumer會與所有的Master和所有的Slave都建立連接進行監聽新消息

五、核心概念

1、Message

消息載體。Message發送或者消費的時候必須指定Topic。Message有一個可選的Tag項用於過濾消息，還可以添加額外的鍵值對。

2、topic

消息的邏輯分類，發消息之前必須要指定一個topic才能發，就是將這條消息發送到這個topic上。消費消息的時候指定這個topic進行消費。就是邏輯分類。

3、queue

1個Topic會被分為N個Queue，數量是可配置的。message本身其實是存儲到queue上的，消費者消費的也是queue上的消息。多說一嘴，比如1個topic4個queue，有5個Consumer都在消費這個topic，那麼會有一個consumer浪費掉了，因為負載均衡策略，每個consumer消費1個queue，5>4，溢出1個，這個會不工作。

4、Tag

Tag 是 Topic 的進一步細分，顧名思義，標籤。每個發送的時候消息都能打tag，消費的時候可以根據tag進行過濾，選擇性消費。

5、Message Model

消息模型：集群（Clustering）和廣播（Broadcasting）

6、Message Order

消息順序：順序（Orderly）和併發（Concurrently）

7、Producer Group

消息生產者組

8、Consumer Group

消息消費者組

六、ACK

首先要明確一點：ACK機制是發生在Consumer端的，不是在Producer端的。也就是說Consumer消費完消息后要進行ACK確認，如果未確認則代表是消費失敗，這時候Broker會進行重試策略（僅集群模式會重試）。ACK的意思就是：Consumer說：ok，我消費成功了。這條消息給我標記成已消費吧。

七、消費模式

1、集群模式（Clustering）

1.1、圖解

 

1.2、特點

  • 每條消息只需要被處理一次，broker只會把消息發送給消費集群中的一個消費者

  • 在消息重投時，不能保證路由到同一台機器上

  • 消費狀態由broker維護

2、廣播模式（Broadcasting）

2.1、圖解

 

2.2、特點

  • 消費進度由consumer維護

  • 保證每個消費者都消費一次消息

  • 消費失敗的消息不會重投

八、Java API

說明：

  • RocketMQ服務端版本為目前最新版：4.7.0

  • Java客戶端版本採取的目前最新版：4.7.0

pom如下

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.0</version>
</dependency>

1、Producer

發消息肯定要必備如下幾個條件：

  • 指定生產組名（不能用默認的，會報錯）

  • 配置namesrv地址（必須）

  • 指定topic name（必須）

  • 指定tag/key（可選）

驗證消息是否發送成功：消息發送完后可以啟動消費者進行消費，也可以去管控台上看消息是否存在。

1.1、send（同步）

public class Producer {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();
        // 創建消息對象，topic為：myTopic001，消息內容為：hello world
        Message msg = new Message("myTopic001", "hello world".getBytes());
        // 發送消息到mq，同步的
        SendResult result = producer.send(msg);
        System.out.println("發送消息成功！result is : " + result);
        // 關閉Producer
        producer.shutdown();
        System.out.println("生產者 shutdown！");
    }
}

輸出結果：

發送消息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854140F418B4AAC26F7973910000, offsetMsgId=7B39B49D00002A9F00000000000589BE, messageQueue=MessageQueue [topic=myTopic001, brokerName=broker-a, queueId=0], queueOffset=7]
生產者 shutdown！

1.2、send（批量）

public class ProducerMultiMsg {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        String topic = "myTopic001";
        // 創建消息對象，topic為：myTopic001，消息內容為：hello world1/2/3
        Message msg1 = new Message(topic, "hello world1".getBytes());
        Message msg2 = new Message(topic, "hello world2".getBytes());
        Message msg3 = new Message(topic, "hello world3".getBytes());
        // 創建消息對象的集合，用於批量發送
        List<Message> msgs = new ArrayList<>();
        msgs.add(msg1);
        msgs.add(msg2);
        msgs.add(msg3);
        // 批量發送的api的也是send()，只是他的重載方法支持List<Message>，同樣是同步發送。
        SendResult result = producer.send(msgs);
        System.out.println("發送消息成功！result is : " + result);
        // 關閉Producer
        producer.shutdown();
        System.out.println("生產者 shutdown！");
    }
}

輸出結果：

發送消息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854139C418B4AAC26F7D13770000,A9FE854139C418B4AAC26F7D13770001,A9FE854139C418B4AAC26F7D13770002, offsetMsgId=7B39B49D00002A9F0000000000058A62,7B39B49D00002A9F0000000000058B07,7B39B49D00002A9F0000000000058BAC, messageQueue=MessageQueue [topic=myTopic001, brokerName=broker-a, queueId=0], queueOffset=8]
生產者 shutdown！

從結果中可以看到只有一個msgId，所以可以發現雖然是三條消息對象，但是卻只發送了一次，大大節省了client與server的開銷。

錯誤情況：

批量發送的topic必須是同一個，如果message對象指定不同的topic，那麼批量發送的時候會報錯：

Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: Failed to initiate the MessageBatch
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
    at org.apache.rocketmq.client.producer.DefaultMQProducer.batch(DefaultMQProducer.java:950)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:898)
    at com.chentongwei.mq.rocketmq.ProducerMultiMsg.main(ProducerMultiMsg.java:29)
Caused by: java.lang.UnsupportedOperationException: The topic of the messages in one batch should be the same
    at org.apache.rocketmq.common.message.MessageBatch.generateFromList(MessageBatch.java:58)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.batch(DefaultMQProducer.java:942)
    ... 2 more

1.3、sendCallBack（異步）

public class ProducerASync {
    public static void main(String[] args) throws Exception {
       // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        // 創建消息對象，topic為：myTopic001，消息內容為：hello world async
        Message msg = new Message("myTopic001", "hello world async".getBytes());
        // 進行異步發送，通過SendCallback接口來得知發送的結果
        producer.send(msg, new SendCallback() {
            // 發送成功的回調接口
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("發送消息成功！result is : " + sendResult);
            }
            // 發送失敗的回調接口
            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
                System.out.println("發送消息失敗！result is : " + throwable.getMessage());
            }
        });

        producer.shutdown();
        System.out.println("生產者 shutdown！");
    }
}

輸出結果：

生產者 shutdown！
java.lang.IllegalStateException: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [124.57.180.156:9876] failed
    at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:681)
    at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:511)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.tryToFindTopicPublishInfo(DefaultMQProducerImpl.java:692)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:556)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.access$300(DefaultMQProducerImpl.java:97)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl$4.run(DefaultMQProducerImpl.java:510)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [124.57.180.156:9876] failed
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.getAndCreateNameserverChannel(NettyRemotingClient.java:441)
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.getAndCreateChannel(NettyRemotingClient.java:396)
    at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:365)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1371)
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1361)
    at org.apache.rocketmq.client.impl.factory.MQClientInstance.updateTopicRouteInfoFromNameServer(MQClientInstance.java:624)
    ... 10 more
發送消息失敗！result is : org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to [124.57.180.156:9876] failed

為啥報錯了？很簡單，他是異步的，從結果就能看出來，由於是異步的，我還沒發送到mq呢，你就先給我shutdown了。肯定不行，所以我們在shutdown前面sleep 1s在看效果

public class ProducerASync {
    public static void main(String[] args) throws Exception {
       // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        // 創建消息對象，topic為：myTopic001，消息內容為：hello world async
        Message msg = new Message("myTopic001", "hello world async".getBytes());
        // 進行異步發送，通過SendCallback接口來得知發送的結果
        producer.send(msg, new SendCallback() {
            // 發送成功的回調接口
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("發送消息成功！result is : " + sendResult);
            }
            // 發送失敗的回調接口
            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
                System.out.println("發送消息失敗！result is : " + throwable.getMessage());
            }
        });

        Thread.sleep(1000);

        producer.shutdown();
        System.out.println("生產者 shutdown！");
    }
}

輸出結果：

發送消息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854106E418B4AAC26F8719B20000, offsetMsgId=7B39B49D00002A9F0000000000058CFC, messageQueue=MessageQueue [topic=myTopic001, brokerName=broker-a, queueId=1], queueOffset=2]
生產者 shutdown！

1.4、sendOneway

public class ProducerOneWay {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();

        // 創建消息對象，topic為：myTopic001，消息內容為：hello world oneway
        Message msg = new Message("myTopic001", "hello world oneway".getBytes());
        // 效率最高，因為oneway不關心是否發送成功，我就投遞一下我就不管了。所以返回是void
        producer.sendOneway(msg);
        System.out.println("投遞消息成功！，注意這裡是投遞成功，而不是發送消息成功哦！因為我sendOneway也不知道到底成沒成功，我沒返回值的。");
        producer.shutdown();
        System.out.println("生產者 shutdown！");
    }
}

輸出結果：

投遞消息成功！，注意這裡是投遞成功，而不是發送消息成功哦！因為我sendOneway也不知道到底成沒成功，我沒返回值的。
生產者 shutdown！

1.5、效率對比

sendOneway > sendCallBack > send批量 > send單條

很容易理解，sendOneway不求結果，我就負責投遞，我不管你失敗還是成功，相當於中轉站，來了我就扔出去，我不進行任何其他處理。所以最快。

而sendCallBack是異步發送肯定比同步的效率高。

send批量和send單條的效率也是分情況的，如果只有1條msg要發，那還搞毛批量，直接send單條完事。

2、Consumer

每個consumer只能關注一個topic。

發消息肯定要必備如下幾個條件：

  • 指定消費組名（不能用默認的，會報錯）

  • 配置namesrv地址（必須）

  • 指定topic name（必須）

  • 指定tag/key（可選）

2.1、CLUSTERING

集群模式，默認。

比如啟動五個Consumer，Producer生產一條消息后，Broker會選擇五個Consumer中的其中一個進行消費這條消息，所以他屬於點對點消費模式。

public class Consumer {
    public static void main(String[] args) throws Exception {
        // 指定消費組名為my-consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my-consumer");
        // 配置namesrv地址
        consumer.setNamesrvAddr("124.57.180.156:9876");
        // 訂閱topic：myTopic001 下的全部消息（因為是*，*指定的是tag標籤，代表全部消息，不進行任何過濾）
        consumer.subscribe("myTopic001", "*");
        // 註冊監聽器，進行消息消息。
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : msgs) {
                    String str = new String(msg.getBody());
                    // 輸出消息內容
                    System.out.println(str);
                }
                // 默認情況下，這條消息只會被一個consumer消費，這叫點對點消費模式。也就是集群模式。
                // ack確認
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 啟動消費者
        consumer.start();
        System.out.println("Consumer start");
    }
}

2.2、BROADCASTING

廣播模式。

比如啟動五個Consumer，Producer生產一條消息后，Broker會把這條消息廣播到五個Consumer中，這五個Consumer分別消費一次，每個都消費一次。

// 代碼里只需要添加如下這句話即可：
consumer.setMessageModel(MessageModel.BROADCASTING); 

2.3、兩種模式對比

  • 集群默認是默認的，廣播模式是需要手動配置。

  • 一條消息：集群模式下的多個Consumer只會有一個Consumer消費。廣播模式下的每一個Consumer都會消費這條消息。

  • 廣播模式下，發送一條消息后，會被當前被廣播的所有Consumer消費，但是後面新加入的Consumer不會消費這條消息，很好理解：村裡面大喇叭喊了全村來領雞蛋，第二天你們村新來個人，那個人肯定聽不到昨天大喇叭喊的消息呀。

3、TAG&&KEY

發送/消費 消息的時候可以指定tag/key來進行過濾消息，支持通配符。*代表消費此topic下的全部消息，不進行過濾。

看下org.apache.rocketmq.common.message.Message源碼可以發現發消息的時候可以指定tag和keys：

public Message(String topic, String tags, String keys, byte[] body) {
    this(topic, tags, keys, 0, body, true);
}

比如：

public class ProducerTagsKeys {
    public static void main(String[] args) throws Exception {
        // 指定生產組名為my-producer
        DefaultMQProducer producer = new DefaultMQProducer("my-producer");
        // 配置namesrv地址
        producer.setNamesrvAddr("124.57.180.156:9876");
        // 啟動Producer
        producer.start();
        // 創建消息對象，topic為：myTopic001，消息內容為：hello world，且tags為：test-tags，keys為test-keys
        Message msg = new Message("myTopic001", "test-tags", "test-keys", "hello world".getBytes());
        // 發送消息到mq，同步的
        SendResult result = producer.send(msg);
        System.out.println("發送消息成功！result is : " + result);
        // 關閉Producer
        producer.shutdown();
        System.out.println("生產者 shutdown！");
    }
}

輸出結果：

發送消息成功！result is : SendResult [sendStatus=SEND_OK, msgId=A9FE854149DC18B4AAC26FA4B7200000, offsetMsgId=7B39B49D00002A9F0000000000058DA6, messageQueue=MessageQueue [topic=myTopic001, brokerName=broker-a, queueId=3], queueOffset=3]
生產者 shutdown！

查看管控台，可以發現tags和keys已經生效了：

 

消費的時候如果指定*那就是此topic下的全部消息，我們可以指定前綴通配符，比如：

// 這樣就只會消費myTopic001下的tag為test-*開頭的消息。
consumer.subscribe("myTopic001", "test-*");

// 代表訂閱Topic為myTopic001下的tag為TagA或TagB的所有消息
consumer.subscribe("myTopic001", "TagA||TagB");

還支持SQL表達式過濾，不是很常用。不BB了。

4、常見錯誤

4.1、sendDefaultImpl call timeout

4.1.1、異常

Exception in thread "main" org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:666)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1342)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1288)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:324)
    at com.chentongwei.mq.rocketmq.Producer.main(Producer.java:18)

4.1.2、解決

1.如果你是雲服務器，首先檢查安全組是否允許9876這個端口訪問，是否開啟了防火牆，如果開啟了的話是否將9876映射了出去。

2.修改配置文件broker.conf，加上：

brokerIP1=我用的是阿里雲服務器，這裡是我的公網IP

啟動namesrv和broker的時候加上本機IP（我用的是阿里雲服務器，這裡是我的公網IP）：

./bin/mqnamesrv -n IP:9876
./bin/mqbroker -n IP:9876 -c conf/broker.conf

4.2、No route info of this topic

4.2.1、異常

Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No route info of this topic: myTopic001
See http://rocketmq.apache.org/docs/faq/ for further details.
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:684)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1342)
    at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1288)
    at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:324)
    at com.chentongwei.mq.rocketmq.Producer.main(Producer.java:18)

4.2.2、解決

很明顯發送成功了，不再是剛才的超時了，但是告訴我們沒有這個topic。那不能每次都手動創建呀，所以啟動broker的時候可以指定參數讓broker為我們自動創建。如下

./bin/mqbroker -n IP:9876 -c conf/broker.conf autoCreateTopicEnable=true

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

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※幫你省時又省力,新北清潔一流服務好口碑

※回頭車貨運收費標準