前言

上一篇【.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深入淺出系列》

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

【其他文章推薦】

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

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

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

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

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

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