前言

上一篇【.Net Core微服務入門全紀錄（五）——Ocelot-API網關（下）】中已經完成了Ocelot + Consul的搭建，這一篇簡單說一下EventBus。

EventBus-事件總線

• 首先，什麼是事件總線呢？

貼一段引用：

事件總線是對觀察者（發布-訂閱）模式的一種實現。它是一種集中式事件處理機制，允許不同的組件之間進行彼此通信而又不需要相互依賴，達到一種解耦的目的。

如果沒有接觸過EventBus，可能不太好理解。其實EventBus在客戶端開發中應用非常廣泛（android，ios，web前端等），用於多個組件（或者界面）之間的相互通信，懂的人都懂。。。

• 那麼，我們為什麼要用EventBus呢？

就拿當前的項目舉例，我們有一個訂單服務，一個產品服務。客戶端有一個下單功能，當用戶下單時，調用訂單服務的下單接口，那麼下單接口需要調用產品服務的減庫存接口，這涉及到服務與服務之間的調用。那麼服務之間又怎麼調用呢？直接RESTAPI？或者效率更高的gRPC？可能這兩者各有各的使用場景，但是他們都存在一個服務之間的耦合問題，或者難以做到異步調用。

試想一下：假設我們下單時調用訂單服務，訂單服務需要調用產品服務，產品服務又要調用物流服務，物流服務再去調用xx服務 等等。。。如果每個服務處理時間需要2s，不使用異步的話，那這種體驗可想而知。

如果使用EventBus的話，那麼訂單服務只需要向EventBus發一個“下單事件”就可以了。產品服務會訂閱“下單事件”，當產品服務收到下單事件時，自己去減庫存就好了。這樣就避免了兩個服務之間直接調用的耦合性，並且真正做到了異步調用。

既然涉及到多個服務之間的異步調用，那麼就不得不提分佈式事務。分佈式事務並不是微服務獨有的問題，而是所有的分佈式系統都會存在的問題。
關於分佈式事務，可以查一下“CAP原則”和“BASE理論”了解更多。當今的分佈式系統更多的會追求事務的最終一致性。

下面使用國人開發的優秀項目“CAP”，來演示一下EventBus的基本使用。之所以使用“CAP”是因為它既能解決分佈式系統的最終一致性，同時又是一個EventBus，它具備EventBus的所有功能！
作者介紹：https://www.cnblogs.com/savorboard/p/cap.html

CAP使用

• 環境準備

在Docker中準備一下需要的環境，首先是數據庫，數據庫我使用PostgreSQL，用別的也行。CAP支持：SqlServer，MySql，PostgreSql，MongoDB。
關於在Docker中運行PostgreSQL可以看我的另一篇博客：https://www.cnblogs.com/xhznl/p/13155054.html

然後是MQ，這裏我使用RabbitMQ，Kafka也可以。
Docker運行RabbitMQ：

docker pull rabbitmq:management
docker run -d -p 15672:15672 -p 5672:5672 --name rabbitmq rabbitmq:management

默認用戶：guest，密碼：guest

環境準備就完成了，Docker就是這麼方便。。。

• 代碼修改：

為了模擬以上業務，需要修改大量代碼，下面代碼如有遺漏的直接去github找。

NuGet安裝：

Microsoft.EntityFrameworkCore
Microsoft.EntityFrameworkCore.Tools
Npgsql.EntityFrameworkCore.PostgreSQL

CAP相關：

DotNetCore.CAP
DotNetCore.CAP.RabbitMQ
DotNetCore.CAP.PostgreSql

Order.API/Controllers/OrdersController.cs增加下單接口：

[Route("[controller]")]
[ApiController]
public class OrdersController : ControllerBase
{
    private readonly ILogger<OrdersController> _logger;
    private readonly IConfiguration _configuration;
    private readonly ICapPublisher _capBus;
    private readonly OrderContext _context;

    public OrdersController(ILogger<OrdersController> logger, IConfiguration configuration, ICapPublisher capPublisher, OrderContext context)
    {
        _logger = logger;
        _configuration = configuration;
        _capBus = capPublisher;
        _context = context;
    }

    [HttpGet]
    public IActionResult Get()
    {
        string result = $"【訂單服務】{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}——" +
            $"{Request.HttpContext.Connection.LocalIpAddress}:{_configuration["ConsulSetting:ServicePort"]}";
        return Ok(result);
    }

    /// <summary>
    /// 下單 發布下單事件
    /// </summary>
    /// <param name="order"></param>
    /// <returns></returns>
    [Route("Create")]
    [HttpPost]
    public async Task<IActionResult> CreateOrder(Models.Order order)
    {
        using (var trans = _context.Database.BeginTransaction(_capBus, autoCommit: true))
        {
            //業務代碼
            order.CreateTime = DateTime.Now;
            _context.Orders.Add(order);

            var r = await _context.SaveChangesAsync() > 0;

            if (r)
            {
                //發布下單事件
                await _capBus.PublishAsync("order.services.createorder", new CreateOrderMessageDto() { Count = order.Count, ProductID = order.ProductID });
                return Ok();
            }
            return BadRequest();
        }

    }

}

Order.API/MessageDto/CreateOrderMessageDto.cs：

/// <summary>
/// 下單事件消息
/// </summary>
public class CreateOrderMessageDto
{
    /// <summary>
    /// 產品ID
    /// </summary>
    public int ProductID { get; set; }

    /// <summary>
    /// 購買數量
    /// </summary>
    public int Count { get; set; }
}

Order.API/Models/Order.cs訂單實體類：

public class Order
{
    [Key]
    [DatabaseGenerated(DatabaseGeneratedOption.Identity)]
    public int ID { get; set; }

    /// <summary>
    /// 下單時間
    /// </summary>
    [Required]
    public DateTime CreateTime { get; set; }

    /// <summary>
    /// 產品ID
    /// </summary>
    [Required]
    public int ProductID { get; set; }

    /// <summary>
    /// 購買數量
    /// </summary>
    [Required]
    public int Count { get; set; }
}

Order.API/Models/OrderContext.cs數據庫Context：

public class OrderContext : DbContext
{
    public OrderContext(DbContextOptions<OrderContext> options)
       : base(options)
    {

    }

    public DbSet<Order> Orders { get; set; }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {

    }
}

Order.API/appsettings.json增加數據庫連接字符串：

"ConnectionStrings": {
  "OrderContext": "User ID=postgres;Password=pg123456;Host=host.docker.internal;Port=5432;Database=Order;Pooling=true;"
}

Order.API/Startup.cs修改ConfigureServices方法，添加Cap配置：

public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers();

    services.AddDbContext<OrderContext>(opt => opt.UseNpgsql(Configuration.GetConnectionString("OrderContext")));

    //CAP
    services.AddCap(x =>
    {
        x.UseEntityFramework<OrderContext>();

        x.UseRabbitMQ("host.docker.internal");
    });
}

以上是訂單服務的修改。

Product.API/Controllers/ProductsController.cs增加減庫存接口：

[Route("[controller]")]
[ApiController]
public class ProductsController : ControllerBase
{
    private readonly ILogger<ProductsController> _logger;
    private readonly IConfiguration _configuration;
    private readonly ICapPublisher _capBus;
    private readonly ProductContext _context;

    public ProductsController(ILogger<ProductsController> logger, IConfiguration configuration, ICapPublisher capPublisher, ProductContext context)
    {
        _logger = logger;
        _configuration = configuration;
        _capBus = capPublisher;
        _context = context;
    }

    [HttpGet]
    public IActionResult Get()
    {
        string result = $"【產品服務】{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}——" +
            $"{Request.HttpContext.Connection.LocalIpAddress}:{_configuration["ConsulSetting:ServicePort"]}";
        return Ok(result);
    }

    /// <summary>
    /// 減庫存 訂閱下單事件
    /// </summary>
    /// <param name="message"></param>
    /// <returns></returns>
    [NonAction]
    [CapSubscribe("order.services.createorder")]
    public async Task ReduceStock(CreateOrderMessageDto message)
    {
        //業務代碼
        var product = await _context.Products.FirstOrDefaultAsync(p => p.ID == message.ProductID);
        product.Stock -= message.Count;

        await _context.SaveChangesAsync();
    }

}

Product.API/MessageDto/CreateOrderMessageDto.cs：

/// <summary>
/// 下單事件消息
/// </summary>
public class CreateOrderMessageDto
{
    /// <summary>
    /// 產品ID
    /// </summary>
    public int ProductID { get; set; }

    /// <summary>
    /// 購買數量
    /// </summary>
    public int Count { get; set; }
}

Product.API/Models/Product.cs產品實體類：

public class Product
{
    [Key]
    [DatabaseGenerated(DatabaseGeneratedOption.Identity)]
    public int ID { get; set; }

    /// <summary>
    /// 產品名稱
    /// </summary>
    [Required]
    [Column(TypeName = "VARCHAR(16)")]
    public string Name { get; set; }

    /// <summary>
    /// 庫存
    /// </summary>
    [Required]
    public int Stock { get; set; }
}

Product.API/Models/ProductContext.cs數據庫Context：

public class ProductContext : DbContext
{
    public ProductContext(DbContextOptions<ProductContext> options)
       : base(options)
    {

    }

    public DbSet<Product> Products { get; set; }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);
        
        //初始化種子數據
        modelBuilder.Entity<Product>().HasData(new Product
        {
            ID = 1,
            Name = "產品1",
            Stock = 100
        },
        new Product
        {
            ID = 2,
            Name = "產品2",
            Stock = 100
        });
    }
}

Product.API/appsettings.json增加數據庫連接字符串：

"ConnectionStrings": {
  "ProductContext": "User ID=postgres;Password=pg123456;Host=host.docker.internal;Port=5432;Database=Product;Pooling=true;"
}

Product.API/Startup.cs修改ConfigureServices方法，添加Cap配置：

public void ConfigureServices(IServiceCollection services)
{
    services.AddControllers();

    services.AddDbContext<ProductContext>(opt => opt.UseNpgsql(Configuration.GetConnectionString("ProductContext")));

    //CAP
    services.AddCap(x =>
    {
        x.UseEntityFramework<ProductContext>();

        x.UseRabbitMQ("host.docker.internal");
    });
}

以上是產品服務的修改。

訂單服務和產品服務的修改到此就完成了，看着修改很多，其實功能很簡單。就是各自增加了自己的數據庫表，然後訂單服務增加了下單接口，下單接口會發出“下單事件”。產品服務增加了減庫存接口，減庫存接口會訂閱“下單事件”。然後客戶端調用下單接口下單時，產品服務會減去相應的庫存，功能就這麼簡單。

關於EF數據庫遷移之類的基本使用就不介紹了。使用Docker重新構建鏡像，運行訂單服務，產品服務：

docker build -t orderapi:1.1 -f ./Order.API/Dockerfile .
docker run -d -p 9060:80 --name orderservice orderapi:1.1 --ConsulSetting:ServicePort="9060"
docker run -d -p 9061:80 --name orderservice1 orderapi:1.1 --ConsulSetting:ServicePort="9061"
docker run -d -p 9062:80 --name orderservice2 orderapi:1.1 --ConsulSetting:ServicePort="9062"

docker build -t productapi:1.1 -f ./Product.API/Dockerfile .
docker run -d -p 9050:80 --name productservice productapi:1.1 --ConsulSetting:ServicePort="9050"
docker run -d -p 9051:80 --name productservice1 productapi:1.1 --ConsulSetting:ServicePort="9051"
docker run -d -p 9052:80 --name productservice2 productapi:1.1 --ConsulSetting:ServicePort="9052"

最後 Ocelot.APIGateway/ocelot.json 增加一條路由配置：

好了，進行到這裏，整個環境就有點複雜了。確保我們的PostgreSQL，RabbitMQ，Consul，Gateway，服務實例都正常運行。

服務實例運行成功后，數據庫應該是這樣的：

產品表種子數據：

cap.published表和cap.received表是由CAP自動生成的，它內部是使用本地消息表+MQ來實現異步確保。

運行測試

這次使用Postman作為客戶端調用下單接口（9070是之前的Ocelot網關端口）：

訂單庫published表：

訂單庫order表：

產品庫received表：

產品庫product表：

再試一下：

OK，完成。雖然功能很簡單，但是我們實現了服務的解耦，異步調用，和最終一致性。

總結

注意，上面的例子純粹是為了說明EventBus的使用，實際中的下單流程絕對不會這麼做的！希望大家不要較真。。。

可能有人會說如果下單成功，但是庫存不足導致減庫存失敗了怎麼辦，是不是要回滾訂單表的數據？如果產生這種想法，說明還沒有真正理解最終一致性的思想。首先下單前肯定會檢查一下庫存數量，既然允許下單那麼必然是庫存充足的。這裏的事務是指：訂單保存到數據庫，和下單事件保存到cap.published表（保存到cap.published表理論上就能夠發送到MQ）這兩件事情，要麼一同成功，要麼一同失敗。如果這個事務成功，那麼就可以認為這個業務流程是成功的，至於產品服務的減庫存是否成功那就是產品服務的事情了（理論上也應該是成功的，因為消息已經確保發到了MQ，產品服務必然會收到消息），CAP也提供了失敗重試，和失敗回調機制。

如果非要數據回滾也是能實現的，CAP的ICapPublisher.Publish方法提供一個callbackName參數，當減庫存時，可以觸發這個回調。其本質也是通過發布訂閱完成，這是不推薦的做法，就不詳細說了，有興趣自己研究一下。
另外，CAP無法保證消息不重複，實際使用中需要自己考慮一下消息的重複過濾和冪等性。

這一篇內容有點多，不知道有沒有表達清楚，有問題歡迎評論交流，如有不對之處還望大家指出。

下一篇計劃寫一下授權認證相關的內容。

代碼放在：https://github.com/xiajingren/NetCoreMicroserviceDemo

未完待續…

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

【其他文章推薦】

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

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

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

※南投搬家公司費用,距離,噸數怎麼算?達人教你簡易估價知識!

※教你寫出一流的銷售文案?

※超省錢租車方案

public class ArrayList {
    private Object[] elementData;
    public Object get(int i) {....}
    public void add(Object o) {....}
}

ArrayList files = new ArrayList();
files.add(new File(""));
String filename = (String)files.get(0);

ArrayList<String> files = new ArrayList<>();

String filename = files.get(0);

files.add(new File(""));

public class Couple<T> {
   private T one;
   private T two;
}

public class Couple<T, U> {...}

//域
private T one;
//返回類型
public T getOne() { return one; }
//局部變量
public void setOne(T newValue) { one = newValue; }

Couple<Rapper>

class ArrayUtil {

    public static <T> T getMiddle(T...a){
        return a[a.length / 2];
    }
}

String middle = ArrayUtil.<String>getMiddle("a","b","c");

String middle = ArrayAlg.getMiddle("a","b","c");

<T extends Serializable & Cloneable>

<T extends Serializable,Cloneable>

Rapper[] rappers = ...;
Singer[] singer = rappers;

Couple<? extends Singer>

public static void printCps(Couple<Rapper> cps) {
      Rapper one = cp.getOne();
      Rapper two = cp.getTwo();
      System.out.println(one.getName() + " & " + two.getName() + " are cps.");
}

public static void printCps(Couple< ? extends Singer> cps) 

Couple<Rapper> rapper = new Couple<>(rapper1, rapper2);
Couple<? extends Singer> singer = rapper;
player.setOne(reader);

? extends Singer getOne();
void setOne(? extends Singer);

? super Rapper

Couple<Singer> cps = ...;
Singer one = cp.getOne();

//我們寫的代碼
Singer one = cps.one;
//編譯器做的事情
Singer one = (Singer)cps.one;

public static <T extends Comparable> T min(T[] a)

public static Comparable min(Comparable[] a)

class Coupling extends Couple<People> {
    public void setTwo(People people) {
            super.setTwo(people);
    }
}

class Coupling extends Couple {
    public void setTwo(People People) {...}
}

public void setTwo(Object two)

Coupling coupling = new Coupling(...);
Couple<People> cp = interval;
cp.setTwo(people);

public void setTwo(Object second) {
    setTwo((People)second);
}

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

【其他文章推薦】

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

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

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

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

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

※網頁設計最專業,超強功能平台可客製化

最近又要用到rabbitmq，業務上要求服務器只發一次消息，需要多個客戶端都去單獨消費。但我們知道rabbitmq的機制里，每個隊列里的消息只能消費一次，所以客戶端要單獨消費信息，就必須得每個客戶端單獨監聽一個queue。所以我最終想實現的是服務端只聲明exchange，客戶端來創建queue和綁定exchange。但是在看各種rabbitmq博文和討論的時候，我覺得對exchange的模式和queue間的關係講的都不是很清楚。所以我決定自己驗證一下

fanout模式和direct模式

本文主要驗證fanout模式和direct模式下以上猜想是否可行。fanout模式就是大名鼎鼎的廣播模式了，只要queue綁定了fanout的交換器，就可以直接的收到消息，無需routingkey的參与。而direct模式就是通過routing key直接發送到綁定了同樣routing key的隊列中。那麼，在這兩種exchange的模式下，是否都可以實現服務端僅創建exchange，客戶端創建queue並綁定exchange呢？

Direct模式驗證

我們先把交換器、routingkey、隊列的名稱定義好:

1. 交換器為directTest
2. routingkey為direct_routing_key
3. 隊列測試3個，首先測試Direct_test_queue_1，再行測試Direct_test_queue_2，再行測試Direct_test_queue_3

代碼使用spring boot框架快速搭建。我們先規劃好需要幾個類來完成這個事情：

1. 針對生產者，需要RabbitmqConfig，用來配置exchange的
2. 針對生產者，需要DirectRabbitSender，用來實現Direct模式的消息發送
3. 針對消費者，需要DirectConsumerOne，來測試第一個隊列Direct_test_queue_1生成和消息接收
4. 針對消費者，需要DirectConsumerTwo，來測試第二個隊列Direct_test_queue_2生成和消息接收
5. 針對消費者，需要DirectConsumerThree，來測試第三個隊列Direct_test_queue_3生成和消息接收
6. 我們還需要一個測試類RabbitmqApplicationTests，用於測試消息的發送和接收

rabbitmq先配置一個DirectExchange

@Bean
DirectExchange directExchange(){
    return new DirectExchange("directTest", true, false);
}

我們可以看到Direct交換器的名稱定義為了directTest，這時候還未綁定任何的隊列。啟動程序，若我們的設想沒錯，則rabbitmq中應該已經生成了directTest的exchange。

Bingo！directTest交換器成功創建。接下來，我們去編寫DirectRabbitSender的代碼

@Component
public class DirectRabbitSender{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private final String EXCHANGE_NAME = "directTest";
    private final String ROUTING_KEY = "direct_routing_key";

    public void send(Object message) {
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, message);
    }

}

我們可以看到代碼中，通過rabbitTemplate發送消息到了交換器為directTest，routingkey為direct_routing_key的地方。但這時候我們沒有任何隊列了，自然接不到消息。現在我們去編寫第一個消費者DirectConsumerOne來接受消息。

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "Direct_test_queue_1", durable = "true"),
        exchange = @Exchange(value = "directTest"),
        key = "direct_routing_key"
))
public class DirectConsumerOne {

    @RabbitHandler
    private void onMessage(String message){
        System.out.println("監聽隊列Direct_test_queue_1接到消息" + message);
    }

}

通過代碼可以看到，我們通過@QueueBinding把Direct_test_queue_1隊列綁定到了directTest和direct_routing_key上。Direct_test_queue_1並沒有在rabbitmq創建，這並沒有關係。一般來說，@RabbitListener會自動去創建隊列。啟動程序，我們去看一下rabbitmq里隊列是不是創建了。

Bingo！再次驗證成功。我們去看看綁定關係是不是正確。這時候Direct_test_queue_1應該綁定到了名為directTest的交換器，而綁定的routingkey為direct_routing_key

biubiubiu！綁定關係完全正確。到了這裏，我們進行最後一步，寫了單元測試去發送消息，查看控制台中消費者是否成功收到消息。RabbitmqApplicationTests的代碼如下：

@SpringBootTest
class RabbitmqApplicationTests {

    @Autowired
    private DirectRabbitSender directRabbitSender;

    @Test
    void contextLoads() {
    }

    @Test
    public void directSendTest(){
        directRabbitSender.send("direct-sender");
        directRabbitSender.send("direct-sender_test");
    }

}

啟動測試類，然後去查看控制台。

沒錯，這就是我們想要達到的效果！基本可以宣布Direct模式驗證成功。服務端生成exchange，客戶端去生成隊列綁定的方式在direct模式下完全可行。為了保險起見，再驗證一下生成多個消費者綁定到同一個隊列是否可行。

DirectConsumerTwo代碼如下：

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "Direct_test_queue_2", durable = "true"),
        exchange = @Exchange(value = "directTest"),
        key = "direct_routing_key"
))
public class DirectConsumerTwo {

    @RabbitHandler
    private void onMessage(String message){
        System.out.println("監聽隊列Direct_test_queue_2接到消息" + message);
    }

}

DirectConsumerThree代碼如下：

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "Direct_test_queue_3", durable = "true"),
        exchange = @Exchange(value = "directTest"),
        key = "direct_routing_key"
))
public class DirectConsumerThree {

    @RabbitHandler
    private void onMessage(String message){
        System.out.println("監聽隊列Direct_test_queue_3接到消息" + message);
    }

}

啟動測試類，我們去看兩個地方：

1. rabbitmq是否創建了客戶端綁定的三個隊列Direct_test_queue_1、Direct_test_queue_2、Direct_test_queue_3
2. 消費者應該各自收到2條消息（Test中發送了兩條，參看上面 RabbitmqApplicationTests 的代碼）。那3個隊列，控制台中應該打印了6條消息。

hohohoho！創建成功，並且綁定關係我看了也全都正確。我們去看控制台

6條！沒有任何毛病，至此，可以宣布Direct模式下，完全支持我們最初的想法：服務端生成exchange，客戶端去生成隊列綁定的方式在direct模式下完全可行。

fanout模式驗證

接下來我們驗證一下fanout的方式，基本操作流程和Direct模式一致。代碼的結構也差不多：

1. 針對生產者，需要RabbitmqConfig，直接在Direct模式下的rabbitmqConfig里直接添加Fanout的交換器配置
2. 針對生產者，需要FanoutRabbitSender，用來實現Fanout模式的消息發送
3. 針對消費者，需要FanoutConsumerOne，來測試第一個隊列Fanout_test_queue_1生成和消息接收
4. 針對消費者，需要FanoutConsumerTwo，來測試第二個隊列Fanout_test_queue_2生成和消息接收
5. 針對消費者，需要FanoutConsumerThree，來測試第三個隊列Fanout_test_queue_3生成和消息接收
6. 測試類RabbitmqApplicationTests也直接復用Direact模式下測試的類

我就不多BB，直接上代碼了。

RabbitmqConfig代碼如下

@Configuration
public class RabbitmqConfig {

    @Bean
    DirectExchange directExchange(){
        return new DirectExchange("directTest", true, false);
    }

    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanoutTest", true, false);
    }

}

FanoutRabbitSender的代碼如下，此處和direct模式的區別是Fanout中沒有routingkey，所以代碼里也沒定義routingkey：

@Component
public class FanoutRabbitSender{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private final String EXCHANGE_NAME = "fanoutTest";

    public void send(Object message) {
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, null, message);
    }

}

我們到這裏先啟動程序試試，看看fanoutTest的交換器在沒有綁定隊列的情況下是否生成了。

棒棒棒！和我們想的一樣，那接下來去寫完所有的消費者，這裏和Direct模式最重要的區別是@Exchange中必須要指定type為fanout。direct模式的代碼里沒指定是因為@Exchange的type默認值就是direct。我直接上代碼了：

/**
 * 監聽器主動去聲明queue=fanout_test_queue_1，並綁定到fanoutTest交換器
 */
@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "fanout_test_queue_1", durable = "true"),
        exchange = @Exchange(value = "fanoutTest", type = ExchangeTypes.FANOUT)
))
public class FanoutConsumerOne {

    @RabbitHandler
    private void onMessage(String message){
        System.out.println("監聽隊列fanout_test_queue_1接到消息" + message);
    }

}

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "fanout_test_queue_2", durable = "true"),
        exchange = @Exchange(value = "fanoutTest", type = ExchangeTypes.FANOUT)
))
public class FanoutConsumerTwo {

    @RabbitHandler
    private void onMessage(String message){
        System.out.println("監聽隊列fanout_test_queue_2接到消息" + message);
    }

}

@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "fanout_test_queue_3", durable = "true"),
        exchange = @Exchange(value = "fanoutTest", type = ExchangeTypes.FANOUT)
))
public class FanoutConsumerThree {

    @RabbitHandler
    private void onMessage(String message){
        System.out.println("監聽隊列fanout_test_queue_3接到消息" + message);
    }

}

接着去測試類RabbitmqApplicationTests中加上fanout的發送測試，然後註釋掉direct的單元測試，以便一會造成干擾

@SpringBootTest
class RabbitmqApplicationTests {

    @Autowired
    private DirectRabbitSender directRabbitSender;

    @Autowired
    private FanoutRabbitSender fanoutRabbitSender;

    @Test
    void contextLoads() {
    }

//    @Test
//    public void directSendTest(){
//        directRabbitSender.send("direct-sender");
//        directRabbitSender.send("direct-sender_test");
//    }

    @Test
    public void fanoutSendTest(){
        fanoutRabbitSender.send("fanout-sender_1");
        fanoutRabbitSender.send("fanout-sender_2");
    }

}

代碼都完成了，現在我們啟動測試類，看看控制台是否正常收到了消息

看圖看圖，fanout模式下也完全認證成功！！！那我們可以宣布，文章開頭的猜想完全可以實現。

總結

服務端只聲明exchange，客戶端來創建queue和綁定exchange的方式完全可行。並且在Direct和Fanout模式下都可行。

那我們可以推測在Header模式的交換器和Topic模式的交換器下應該也大差不差。具體各位可自行驗證，基本流程和上面direct和fanout的流程差不多。

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

【其他文章推薦】

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

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

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

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

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

※教你寫出一流的銷售文案?

MongoDB是一種流行的數據庫，可以在不受任何錶格schema模式的約束下工作。數據以類似JSON的格式存儲，並且可以包含不同類型的數據結構。例如，在同一集合collection 中，我們可以擁有以下兩個文檔document：

{
    id: '4',
    name: 'Mark',
    age: '21',
    addresses : [
        { street: '123 Church St', city: 'Miami', cc: 'USA' },
        { street: '123 Mary Av', city: 'Los Angeles', cc: 'USA' }
    ]
}

{
    id: '15',
    name: 'Robin',
    department: 'New Business',
    example: 'robin@example.com'
}

為了能夠充分利用MongoDB的優勢，您必須了解並遵循一些基本的數據庫設計原則。在講解設計方法之前，我們必須首先了解MongoDB存儲數據的結構。

一、 數據如何存儲在MongoDB中

與傳統的RDBMS關係型數據庫不同，MongoDB並沒有表Table，行row和列column的概念。它將數據存儲在集合collections，文檔documents和字段fields中。下圖說明了與RDBMS類比的結構之間的關係：

二、數據庫設計技巧和竅門

2.1.規範化存儲與非規範化存儲

因為MongoDB使用文檔來存儲數據，所以理解“規範化存儲“”和“非規範化存儲”的概念非常重要。

規範化存儲：-規範化意味着將數據存儲到多個集合collections中，並在它們之間設計關聯關係。數據保存之後，更新數據比較容易。但是在讀取數據的時候，規範化存儲的缺點就顯現出來。如果要從多個集合collections查找數據，則必須執行多個查詢，從而使讀取數據的速度變慢。 （比如：將網頁標題、作者、內容分別存儲到不同的collections中）

非規範化存儲：-這種方式將若干對象數據，以嵌套的方式存儲到單個文檔中。它在讀取數據的時候表現更好，但在寫入時會變慢。這種存儲數據的方式還將佔用更多空間。 （比如：將網頁標題、作者、內容分別存儲到同一個collection中）

所以在兩種存儲數據方式之間進行選擇之前，先評估一下你的應用數據庫的使用方式。

• 如果您有一個不需要頻繁更新的數據，更新的即時一致性不是很重要，但是在讀取時需要良好的性能，那麼非規範化可能是明智的選擇。（比如：我們博客的博文，作者一旦保存之後，幾乎就不在進行頻繁的修改，但是面臨着讀者頻繁的讀取閱讀操作）

• 如果數據庫中的文檔數據需要不斷的更新，並且您希望在寫入時具有良好的性能，那麼您可能需要考慮規範化存儲。（比如：需要頻繁修改數據的業務類系統）

2.2. 一對多關係

與RDBMS相比，在MongoDB中對“一對多”關係建模需要進行更細粒度的設計。許多初學者陷入將文檔數組嵌入父文檔中的陷阱。正如我們在上文中介紹的，知道何時進行規範化存儲或非規範化存儲是非常重要的。因此設計者需要考慮關係的基數是“一個對少數幾個”還是“一個對多個”？每種關係將具有不同的建模方法。

例如：下面“一個對少數幾個”的建模示例。最好的建模方法是在父文檔（persopn）中嵌入幾個（address）：

> db.person.findOne()
{
  name: 'Mark Kornfield',
  ssn: '1223-234-75554',
  addresses : [
     { street: '123 Church St', city: 'Miami', cc: 'USA' },
     { street: '123 Mary Av', city: 'Los Angeles', cc: 'USA' }
  ]
}

在“一個對多個”示例中，我們將考慮設計兩個集合，即產品products集合和零件parts集合。每個零件都有一個“ ObjectID”，該“ ObjectID”將出現在產品集合的引用中。這樣的設計可以讓讀寫性能更高效。

> db.parts.findOne()
{
    _id : ObjectID('AAAA'),
    partno : '1224-dsdf-2215',
    name : 'bearing',
    price: 2.63

> db.products.findOne()
{
    name : 'car',
    manufacturer : 'Ford',
    catalog_number: 2234,
    parts : [     // array of references to Part documents
        ObjectID('AAAA'),    // reference to the bearing above
        ObjectID('F17C'),    // reference to a different Part
        ObjectID('D2AA'),
        // etc
]

2.3.設計模式可視化

儘管MongoDB是schemaless“無模式的”，但仍然存在將集合collections可視化為圖表的方法。能夠查看設計圖，將對您理解和設計MongoDB的方式上產生重大影響。

DbSchema是可以很好地完成可視化設計工作的一個工具。如下圖所示，它將通過讀取集合和文檔來推導架構。此外，您只需單擊就可以修改圖中的對象。在DbSchema中，您還可以為MongoDB創建外鍵，當然僅在本地創建，只用於設計目的。

2.4.智能索引

為了保持數據庫的良好性能，有必要建立智能索引，這將簡化寫入和讀取操作。知道MongoDB的索引優勢和局限性非常重要，MongoDB保留用於排序操作的內存限製為32MB。如果你不使用索引，則排序時數據庫將被迫將所有排序文檔hold在內存裏面，如果達到32M的限制，則數據庫將返回錯誤或空集。

結論

對MongoDB的透徹理解與對數據庫想要實現的目標的清晰了解是良好數據庫設計的秘訣。

歡迎關注我的博客，裏面有很多精品合集

• 本文轉載註明出處（必須帶連接，不能只轉文字）：字母哥博客。

覺得對您有幫助的話，幫我點贊、分享！您的支持是我不竭的創作動力！ 。另外，筆者最近一段時間輸出了如下的精品內容，期待您的關注。

• 《手摸手教你學Spring Boot2.0》
• 《Spring Security-JWT-OAuth2一本通》
• 《實戰前後端分離RBAC權限管理系統》
• 《實戰SpringCloud微服務從青銅到王者》
• 《VUE深入淺出系列》

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

【其他文章推薦】

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

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

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

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

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

※產品缺大量曝光嗎?你需要的是一流包裝設計!

工程效能領域，測試覆蓋率度量總是繞不開的話題，我們也不例外。在七牛雲，我們主要使用go語言構建雲服務，在考慮系統測試覆蓋率時，最早也是通過圍繞原生go test -c -cover的能力來構建。這個方案，筆者還曾在 MTSC2018大會上有過專項分享。其實我們當時已經做了不少自動化工作，能夠針對很多類型的代碼庫，自動插樁服務，自動生成TestMain()等方法，但隨着接入項目越來越多，以及後面使用場景的不斷複雜化，我們發現這套還是有其先天局限，會讓後面越來越難受：

• 程序必須關閉才能收集覆蓋率。如果將這套系統僅定位在收集覆蓋率數據上，這個痛點倒也能忍受。但是如果想進一步做精準測試等方向，就很受局限。
• 因為不想污染被測代碼庫，我們採取了自動化的方式，在編譯階段給每個服務生成類似main_test.go文件。但這種方式，其最難受的地方在於flag的處理，要知道go test命令本身會調用flag.Parse方法，所以這裏需要自動化的修改源碼，保證被測程序的flag定義，要先於go test調用flag.Parse之前。但是，隨着程序自己使用flag姿勢的複雜化，我們發現越來越難有通用方案來處理這些flag，有點難受。
• 受限於go test -c命令的先天缺陷，它會給被測程序注入一些測試專屬的flag，比如-test.coverprofile, -test.timeout等等。這個是最難受的，因為它會破壞被測程序的啟動姿勢。我們知道系統測試面對是完整被測集群，如果你需要專門維護一套測試集群來做覆蓋率收集時，就會顯得非常浪費。好鋼就應該用在刀刃上，在七牛雲，我們倡導極客文化，追求用工程師思維解決重複問題。而作為業務效率部門，我們自己更應該走在前列。

也是因為以上的種種考量，我們內部一直在優化這一套系統，到今天這一版，我們已從架構和實現原理上完成了顛覆，能夠做到無損插樁，運行時分析覆蓋率，當屬非常優雅。

Goc – A Comprehensive Coverage Testing System for The Go Programming Language

一圖勝千言:

使用goc run .的姿勢直接運行被測程序，就能在運行時，通過goc profile命令方便的得到覆蓋率結果。是不是很神奇？是不是很優雅？

這個系統就是goc, 設計上希望完全兼容go命令行工具核心命令(go buld/install/run)。使用體驗上，也希望向go命令行工具靠攏：

以下是goc 1.0版本支持的功能:

系統測試覆蓋率收集方案

有了goc，我們再來看如何收集go語言系統測試覆蓋率。整體比較簡單，大體只需要三步：

• 首先通過goc server命令部署一個服務註冊中心，它將會作為樞紐服務跟所有的被測服務通信。

• 使用goc build --center="<server>" 命令編譯被測程序。goc不會破壞被測程序的啟動方式，所以你可以直接將編譯出的二進制發布到集成測試環境。

• 環境部署好之後，就可以做執行任意的系統測試。而在測試期間，可以在任何時間，通過goc profile --center="<server>"拿到當前被測集群的覆蓋率結果。
  是不是很優雅？

goc 核心原理及未來

goc在設計上，拋棄老的go test -c -cover模式，而是直接與go tool cover工具交互，避免因go test命令引入的一系列弊端。goc同樣沒有選擇自己做插樁，也是考慮go語言的兼容性，以及性能問題，畢竟go tool cover工具，原生採用結構體來定義counter收集器，每個文件都有單獨的結構體，性能相對比較可靠。goc旨在做go語言領域綜合性的覆蓋率工具以及精準測試系統，其還有很長的路要走:

• 基於PR的單測/集測/系統覆蓋率增量分析
• 精準測試方向，有一定的產品化設計體驗，方便研發與測試日常使用
• 擁抱各種CICD系統

當前goc已經開源了，歡迎感興趣的同學，前往代碼倉庫查看詳情並Star支持。當然，我們更歡迎有志之士，能夠參与貢獻，和我們一起構建這個有意思的系統。

最後，父親節快樂！

Contact me ?

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

【其他文章推薦】

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

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

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

※超省錢租車方案

※教你寫出一流的銷售文案?

※網頁設計最專業,超強功能平台可客製化

閱讀本文大概需要 2 分鐘。

大家好，這是 C# 9.0 新特性系列的第 4 篇文章。

熟悉函數式編程的童鞋一定對“只讀”這個詞不陌生。為了保證代碼塊自身的“純潔”，函數式編程是不能隨便“弄髒”外來事物（參數、變量等）的，所以“只讀”對函數式編程非常重要。

為了豐富 C# 對函數式編程支持，較新的 C# 版本引入了一些很有用的新特性。比如 C# 8 中就對 struct 類型的方法增加了 readonly 修飾符支持，被 readonly 修飾的方法是不能修改該方法所在類的屬性的。舉個例子：

public struct FooValue
{
    private int A { get; set; }
    public readonly int IncreaseA()
    {
        A = A + 1; // 報錯
        return A;
    }
}

而 C# 9 又進一步增加了對“只讀”的支持，此次增加了 init-only 屬性和 record 相關特性，下面一一介紹。

Init-only 屬性

我們知道類的屬性有 set 和 get 兩種訪問器，現在 C# 9 增加一種屬性訪問器：init。init 是 set 訪問器的變體，它的作用是使屬性只能在對象初始化的時候對其賦值，之後該屬性就是只讀的，因此叫 init-only 屬性。使用方式如下：

public class Foo
{
    public string PropA { get; init; }
    public string PropB { get; init; }
}

賦值操作：

var foo = new Foo {  PropA = "A", PropB = "B" };
foo.PropA = "AA"; // 報錯，PropA 此時是只讀的！

由於 init 是在初始化階段賦值，所以它可以在類內部修改 readonly 修飾的字段。比如：

public class Foo
{
    private readonly string propA;
    private readonly string propB;

    public string PropA
    {
        get => propA;
        init => propA = (value ?? throw new ArgumentNullException(nameof(propA)));
    }
    public string PropA
    {
        get => propB;
        init => propB = (value ?? throw new ArgumentNullException(nameof(propB)));
    }
}

如果你知道在構造函數中可以對只讀字段/屬性賦值就自然也理解這一點。

記錄 (Record)

做過財務系統的人都知道交易記錄一旦入賬是不能修改的，如果錄入錯誤，就要新錄入一筆負的記錄把之前的紅衝掉，再錄入正確的記錄。應對類似這種只讀記錄的場景，C# 9 引入了 Record（記錄，下文均使用中文的“記錄”）的概念，它用來支持整個對象的只讀特性（即實例化後為只讀）。使用方式如下：

public data class Foo
{
    public string PropA { get; init; }
    public string PropB { get; init; }
}

這裏用了一個 data 關鍵字，表示該類的對象只是純粹的記錄值，它不是可修改的狀態（在函數式編程中，所有的數據修改都是狀態在發生變化）。

上面的太麻煩了，可以這樣簡寫：

public data class Foo
{
    string PropA;
    string PropB;
}

默認屬性都是 public 的，如果實在要改為 private，可以在屬性定義前面加上 private 修飾符。

定位記錄 (Positional Record)

有時候為了初始化更方便，可以定義構造函數來給屬性賦值，初始化時只需要把屬性值按順序傳給構造函數即可，這個操作稱為定位構造(Positional Construction)。同樣，也可以使用解構函數(Deconstructor)來實現屬性的解構，即按照解構函數的參數順序從對象中提取屬性的值，被稱為定位解構(Positional Deconstructor)。實現了定位構造或定位解構的記錄稱為定位記錄(Positional Record)。下面是一個定位記錄的實現：

public data class Foo
{
    string PropA;
    string PropB;
    public Foo(string propA, string propB)
      => (PropA, PropB) = (propA, propB);
    public void Deconstruct(out string propA, out string propB)
      =>  (propA, propB) = (PropA, PropB);
}

這個寫法太麻煩了，可以直接簡寫為：

public data class Foo(string PropA, string PropB);

這樣簡短一句代碼，其內部默認實現了 init-only 自動屬性，且同時為所有屬性定義了構造函數和解構函數。

使用示例：

var foo = new Foo("AA", "BB");  // 構造定位
var (a, b) = foo;               // 解構定位

可以想象，記錄的大部分使用場景，以上簡寫的寫法能滿足需求。若有特殊場景，就不能簡單，需要進行自定義修改其默認行為。

with 表達式

當處理不可變數據時，若要生成不同的狀態，一個常見的場景是在一條舊記錄基礎上拷貝一條新的記錄。比如我們要修改 Foo 對象的 PropA 屬性，我們就要拷貝該對象生成一個新的對象。這個操作在函數式編程中被稱為“非破壞性修改 (non-destructive mutation)”。為了支持記錄這個操作，C# 9 引入了 with 表達式，它可以很方便在一條原有記錄基礎上創建一條新記錄。示例：

var other = foo with { PropA = "AA" };

with 表達式內部其實是通過一個默認的 protected 構造函數來實現的，大致如下：

protected Foo(Foo original)
{
    // 拷貝 original 的所有字段
}

如果默認實現的字段拷貝不符合你的需求，你也可以手動實現這個構造函數。

今天就分享到這裏，敬請期待一下篇關於 C# 9 新特性的文章！

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

【其他文章推薦】

※教你寫出一流的銷售文案?

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

※回頭車貨運收費標準

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

※超省錢租車方案

※產品缺大量曝光嗎?你需要的是一流包裝設計!