RabbitMQ、Oracle with C#

從 RabbitMQ 接收資料，只顯示在終端視窗上似乎沒甚麼意思，接下來我們要把它直接寫入我們熟悉的 Oracle 資料庫。因此我們先要了解 C# 與關連式資料庫的運作方式。

首先要學習如何連接(connection)資料庫，以及關閉與資料庫的連接。如何使用查詢，如何添加和更新紀錄。如果你已了解 Oracle Data Provider for .NET (ODP.NET)的操作，則可以跳過 Oracle 範例，直接到後面的 “整合 RabbitMQ 消費訊息與 Oracle”。

ADO.NET 之前使用 OLEDB 與 ODBC 附帶不同的資料庫提供程式(Data Provider)，一個提供程式用於 SQL Server;另一個提供程式用於 Oracle。OLEDB 技術已不再獲得支援，所以這個提供程式不應該再用於新的應用程式。對於 Oracle 資料庫，微軟提供的程式也不再使用，因為來自 Oracle 所提供的程式能更好的滿足需求。我們這裡要使用的 Oracle Data Provider for .NET (ODP.NET) Core 是 Oracle.ManagedDataAccess.Core。

Oracle 範例

我們要在原先的 amqp-example 解決方案之下再開一個 OracleSample 專案來學習 Oracle 資料庫的操作，切換到你的 amqp-example 解決方案目錄:

dotnet new console --name OracleSample

然後在 OracleSample 專案目錄下安裝 NuGet Oracle 套件。

Oracle Data Provider for .NET Core

cd OracleSample
dotnet add package Oracle.ManagedDataAccess.Core

安裝成功後，你可在 OracleSample 專案目錄下的檔案 OracleSample.csproj 發現新的資料紀錄。

<ItemGroup>
  <PackageReference Include="Oracle.ManagedDataAccess.Core" Version="2.19.31"/>
</ItemGroup>

連線 Connection

我們就可以開始了。首先是資料庫的連線。在 OracleSample 目錄下開一個目錄 src 來放我們的範例，然後在 src 目錄下建立一個新的 C# 程式檔 ConnectionSample.cs。

src/ConnectionSample.cs

using System;
using Oracle.ManagedDataAccess.Client;

namespace OracleSample.src
{
  public class ConnectionSample
  {
    public static void OpenConnection()
    {
      string connectionString = "User Id=xxxx;Password=xxxxxxx;Data Source=10.11.xx.xxx:1522/xxxx.xxx.com.tw;";

      var connection = new OracleConnection(connectionString);
      connection.Open();
      Console.WriteLine("Connection opened");
      Console.WriteLine("Press [enter] to continue");
      Console.ReadLine();
      connection.Close();
    }
  }
}

在第 16 行我們加一個 Console.ReadLine() 讓程式暫停，我們才能從資料庫觀察連線狀況。

回到 OracleSample 專案的根目錄，修改一下 Program.cs 以便測試，記得要將 OracleSample.src 命名空間加進來:

Program.cs

using System;
using OracleSample.src;

namespace OracleSample
{
  class Program
  {
    static void Main(string[] args)
    {
      ConnectionSample.OpenConnection();
    }
  }
}

在專案目錄下開啟一個終端視窗來啟動程式:

dotnet run

讓它暫停在終端機上，到資料庫觀察一下連線:

USERNA  PID  SID MACHINE         SPID     STATUS   SERIAL# MODULE               PROGRAM                      CLIENT_INFO
------ ---- ---- --------------- -------- -------- ------- -------------------- ---------------------------- ------------------
DEMO     53   90 XXX\7x0x0x4xP1  428      INACTIVE    3289 OracleSample.dll     OracleSample.dll

OracleConnection 類實現了 IDisposable 介面，其中包含 Dispose() 方法和 Close() 方法。這兩個方法的功能相同，都是釋放連線。這樣，就可以使用 using 語句來自動關閉資料庫連線。

這個 ConnectionSample 範例使用定義好的連接字符串打開資料庫連接，一旦打開連線就可以對資料庫執行命令，完成後關閉連線。

直接在程式碼設定連接字符串，不是一個很適合的方式，最好是將它設定在一個配置檔案中。在 .NET Core 中，配置文件可以是 JSON 或 XML 格式，或從環境變數中讀取。這範例我們使用 JSON 格式。

首先安裝要兩個 NuGet 套件:

cd OracleSample
dotnet add package Microsoft.Extensions.Configuration
dotnet add package Microsoft.Extensions.Configuration.Json

在 OracleSample 專案根目錄下建立 config.json 配置檔:

config.json

{
  "Data": {
    "DefaultConnection": {
      "ConnectionString": "User Id=demo;Password=demo426;Data Source=10.11.xx.xxx: 1522/lxxx.xxx.com.tw;"
    }
  }
}

然後我們要修改 ConnectionSample.cs 從 config.json 配置檔讀取資料庫連線資訊:

src/ConnectionSample.cs

using System;
using Oracle.ManagedDataAccess.Client;
using System.IO;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Configuration.Json;

namespace OracleSample.src
{
  public class ConnectionSample
  {
    public static void OpenConnection()
    {
      var configurationBuilder = new ConfigurationBuilder()
        .SetBasePath(Directory.GetCurrentDirectory())
        .AddJsonFile("config.json");
      IConfiguration config = configurationBuilder.Build();
      string connectionString = config["Data:DefaultConnection:ConnectionString"];

      var connection = new OracleConnection(connectionString);
      connection.Open();
      Console.WriteLine("Connection opened");
      Console.WriteLine("Press [enter] to continue");
      Console.ReadLine();
      connection.Close();
    }
  }
}

程式碼似乎變的複雜一些，但是這是必要的，這可避免為往後增加一些技術債。

這裡使用 ConfigurationBuilder 創建一個實例，AddJsonFile 擴展方法添加 JSON 文件 config.json，從這個文件讀取配置信息。調用 ConfigurationBuilder 的 Build() 方法，從所添加的配置文件中建構配置，返回一個實現了 IConfiguration 介面的物件。這樣就可以檢索配置值。

Pooling

我們可以在設定連接字符串中直接設定連接池(connection pool)。選項 Pooling 設置為 false，會禁用連接池; 預設是啟用的: Pooling=true。 Min Pool Size 和 Max Pool Size 允許配置池中的連接數。預設情況下，Min Pool Size 的值為 1，Max Pool Size 的值為 100。Connection Lifetime 定義了連線在釋放前在池中保持不活躍狀態的時間。

命令 Command

連線到資料庫後就可以針對資料庫執行命令，就是要在資料庫上執行包含 SQL 語句的文本字符串。命令也可以是一個存儲過程 (Stored Procedure)。

把 SQL 子句作為一個參數傳遞給 Command 類的建構函式，就可以建構一條命令。在 OracleSample/src 目錄下建立一個新的程式檔 CommandSample.cs:

src/CommandSample.cs

using System;
using Oracle.ManagedDataAccess.Client;
using System.IO;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Configuration.Json;

namespace OracleSample.src
{
  public class CommandSample
  {
    public static void CreateCommand()
    {
      using (var connection = new OracleConnection(GetConnectionString()))
      {
         string sql = "SELECT empno, ename, hiredate, sal, deptno From emp";
         var command = new OracleCommand(sql, connection);
         connection.Open();
         //...
      }
    }
    private static string GetConnectionString()
    {
      var configurationBuilder = new ConfigurationBuilder()
        .SetBasePath(Directory.GetCurrentDirectory())
        .AddJsonFile("config.json");
      IConfiguration config = configurationBuilder.Build();
      string connectionString = config["Data:DefaultConnection:ConnectionString"];
      return connectionString;
    }
  }
}

這裡我們將讀取配置檔獨立在一個靜態方法 GetConnectionString() 中，讓程式模組化一些。OracleCommand 命令的建構，也可以透過 OracleConnection 的 CreateCommand( ) 方法，把 SQL 語句賦予 CommandText 屬性:

src/CommandSample.cs

OracleCommand command = connection.CreateCommand();
command.CommandText = sql;

命令通常需要參數。例如，下面的 SQL 語句需要一個 deptno 參數。不要試圖使用字符串連接來建立參數，它經常被用於 SQL 注入(SQL injection)攻擊。相反的，應使用 ODP.NET 的參數特性，OracleParameter 物件會抑制這種攻擊:

src/CommandSample.cs

string sql = "SELECT empno, ename, hiredate, sal, deptno From emp where deptno = :deptno";
var command = new OracleCommand(sql, connection);
command.BindByName = true;
OracleParameter deptno = new OracleParameter("deptno", 10);
command.Parameters.Add(deptno);

定義好命令後，就需要執行它。執行語句有許多方式，這取決於要從命令中返回甚麼數據。OracleCommand 類提供了下述可執行的命令:

• ExecuteReader() : 執行命令，返回一個類型化的 IDataReader。
• ExecuteNonQuery() : 執行命令，但不返回任何結果。
• ExecuteScalar(): 執行命令，返回結果集中第一行第一列的值。

ExecuteReader( ) 方法

我們分別來看看這三個方法，在 src 目錄下產生 ExecuteSample.cs，首先看 ExecuteReader()，ExecuteReader( ) 方法執行命令，並返回一個 DataReader 物件，返回的物件可以用於遍歷返回的紀錄。

src/ExecuteSample.cs

using System;
using System.IO;
using Oracle.ManagedDataAccess.Client;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Configuration.Json;

namespace OracleSample.src
{
  public class ExecuteSample
  {
    public static void ExecuteReader(int deptnoParameter)
    {
      string sql = "SELECT empno, ename, hiredate, sal, comm, deptno From emp where deptno = :deptno order by empno";

      try
      {
        using (var connection = new OracleConnection(GetConnectionString()))
        using (var command = new OracleCommand(sql, connection))
        {
          command.BindByName = true;

          OracleParameter deptnoBind = new OracleParameter("deptno", OracleDbType.Int32);
          deptnoBind.Value = deptnoParameter;
          command.Parameters.Add(deptnoBind);

          connection.Open();
          using (var reader = command.ExecuteReader())
          {
            while (reader.Read())
            {
              short empno = reader.GetInt16(0);
              string ename = reader.GetString(1);
              DateTime hiredate = reader.GetDateTime(2);
              float sal = reader.GetFloat(3);
              float? comm = reader.IsDBNull(4) ? (float?)null : reader.GetFloat(4);
              short deptno = reader.GetInt16(5);

              Console.WriteLine($"{empno,4} {ename,-10} {hiredate:yyyy-MM-dd} {sal,7:F2} {comm,7:F2} {deptno,2}");
            }
          }
        }
      }
      catch (OracleException ex)
      {
        Console.WriteLine(ex.Message);
      }
    }
    private static string GetConnectionString()
    {
      var configurationBuilder = new ConfigurationBuilder()
        .SetBasePath(Directory.GetCurrentDirectory())
        .AddJsonFile("config.json");
      IConfiguration config = configurationBuilder.Build();
      string connectionString = config["Data:DefaultConnection:ConnectionString"];
      return connectionString;
    }
  }
}

這裡建構 OracleParameter 的方式與上個範例有些微的不同，我們定義了 deptno 參數的型態 OracleDbType.Int32。

當調用 OracleCommand 物件的 ExecuteReader() 方法時，返回 OracleDataReader 物件。這裡要注意，OracleDataReader 使用後需要銷毀。

從數據讀取器 reader 中讀取記錄時，在 while 循環中調用 Read() 方法。Read() 方法的第一次調用時會將游標 (cursor) 移動到返回的第一筆記錄上，再次調用時游標定位到下一筆記錄，如果沒有記錄了，Read() 方法就返回 false。

訪問列(column)的值時，調用不同的 GetXXX() 方法讀取欄位的值，傳遞給這些方法的索引對應於用 SQL SELECT 語句檢索的列(column)，因此即使資料庫的結構有變化，該索引也保持不變 (不要用 SELECT *)。在強類型化的 GetXXX() 方法中，需要注意從資料庫返回的 null 值，GetXXX () 方法會拋出一個異常。為了避免異常，使用 C# 條件語句 ?: 和 OracleDataReader.IsDBNull() 方法，檢查值是否為 null，如果是，就把 null 分配給可空的 (Nullable Type) decimal? 變數 comm。

可空類型 Nullable Types

引用類型的變數可以為空 (null)，而值類型的變數不能。把 C# 類型映射到資料庫類型時，這是一個特殊的問題。資料庫中的值可以為 null。C# 的解決方案就是: 可空類型 Nullable Types。可空類型是可以為 null 的值類型。可空類型只需要在類型的後面添加”?”。

在下面的程式碼片段中，x1 是一個普通的 int，x2 是一個可以為空的 int，所以可以把 null 分配給 x2:

int x1 = 1;
int? x2 = null;

修改 OracleSample 專案下的 Program.cs 測試:

Program.cs

static void Main(string[] args)
{
  ExecuteSample.ExecuteReader(10);
}

對於 OracleDataReader，可以不使用 GetXXX 方法，而可以使用無類型的索引器返回一個物件，因此，需要轉換為相應的類型:

src/ExecuteSample.cs

short empno = (short)reader[0];
string ename = (string)reader[1];
DateTime hiredate = (DateTime)reader[2];
float sal = (float)reader[3];
float? comm = reader.IsDBNull(4) ? (float?) null : (float?)reader[4];
short deptno = (short)reader[5];

OracleDataReader 的索引器還允許使用 string 而不是 int 傳遞列名 (column name)。在這些不同的選項中，這是最慢的方法，但它的可讀性最佳，與發出服務調用所需的時間相比，訪問索引器所需的額外時間其實可以忽略不計。

src/ExecuteSample.cs

short empno = (short)reader["empno"];
string ename = (string)reader["ename"];
DateTime hiredate = (DateTime)reader["hiredate"];
float sal = (float)reader["sal"];
float? comm = reader.IsDBNull(4) ? (float?)null : (float?)reader["comm"];
short deptno = (short)reader["deptno"];

強烈型的語言有時在映對資料庫的資料型態實在是很困擾，既然是從資料庫返回的資料，不管它是否為 null，總是會有初始值，我們就直接使用 var 類型推斷。 要記得使用 var 類型推斷，一定要賦予初始值，否則它無法推斷:

src/ExecuteSample.cs

var empno = reader["empno"];
var ename = reader["ename"];
var hiredate = reader["hiredate"];
var sal = reader["sal"];
var comm = reader["comm"];
var deptno = reader["deptno"];

你會不會好奇，當 comm 欄位是 null 時，程式是不是會當掉? 或者變數 hiredate 的資料型態是甚麼? 可以使用 GetType() 方法查看它的類別型態。

src/ExecuteSample.cs

Console.WriteLine($"{empno.GetType()} {ename.GetType()} {hiredate.GetType()} {sal.GetType()} {comm.GetType()} {deptno.GetType()}");

結果應該是:

System.Int16 System.String System.DateTime System.Single System.DBNull System.Int16

ExecuteNonQuery( ) 方法

ExecuteNonQuery() 方法一般用於 UPDATE、INSERT 或 DELETE 語句，其中維一返回值是受影響的紀錄筆數。但如果調用帶輸出參數的存儲過程 (Stored Procedure)，該方法就有返回值。

在 ExecuteSample 類中新增一個靜態方法 ExecuteNonQuery( ):

src/ExecuteSample.cs

public static void ExecuteNonQuery()
{
  var newEmployee = new
  {
    empno = 9588,
    ename = "C#範例",
    job = "ANALYST",
    mgr = 7566,
    hiredate = DateTime.Now,
    sal = 23000,
    comm = 111,
    deptno = 40
  };

  string sql = "INSERT INTO EMP (empno, ename, job, mgr, hiredate, sal, comm, deptno) " +
    "VALUES (:empno, :ename, :job, :mgr, :hiredate, :sal, :comm, :deptno)";

  try
  {
    using (var connection = new OracleConnection(GetConnectionString()))
    using (var command = new OracleCommand(sql, connection))
    {
      command.BindByName = true;

      OracleParameter empnoBind = new OracleParameter("empno", OracleDbType.Int32);
      empnoBind.Value = newEmployee.empno;
      command.Parameters.Add(empnoBind);

      OracleParameter enameBind = new OracleParameter("ename", OracleDbType.Varchar2);
      enameBind.Value = newEmployee.ename;
      command.Parameters.Add(enameBind);

      OracleParameter jobBind = new OracleParameter("job", OracleDbType.Varchar2);
      jobBind.Value = newEmployee.job;
      command.Parameters.Add(jobBind);

      OracleParameter mgrBind = new OracleParameter("mgr", OracleDbType.Int32);
      mgrBind.Value = newEmployee.mgr;
      command.Parameters.Add(mgrBind);

      OracleParameter hiredateBind = new OracleParameter("hiredate", OracleDbType.Date);
      hiredateBind.Value = newEmployee.hiredate;
      command.Parameters.Add(hiredateBind);

      OracleParameter salBind = new OracleParameter("sal", OracleDbType.Decimal);
      salBind.Value = newEmployee.sal;
      command.Parameters.Add(salBind);

      OracleParameter commBind = new OracleParameter("comm", OracleDbType.Decimal);
      commBind.Value = newEmployee.comm;
      command.Parameters.Add(commBind);

      OracleParameter deptnoBind = new OracleParameter("deptno", OracleDbType.Int32);
      deptnoBind.Value = newEmployee.deptno;
      command.Parameters.Add(deptnoBind);

      connection.Open();
      int rows = command.ExecuteNonQuery();
      Console.WriteLine($"{rows} row(s) inserted");
    }
  }
  catch (OracleException ex)
  {
    Console.WriteLine(ex.Message);
  }
}

這裡我們使用匿名類型(Anonymous Types)產生一個 newEmployee 物件。var 與 new 關鍵字一起使用時，可以創建匿名類型。匿名類型只是一個繼承自 Object 且沒有名稱的類。該類的定義從物件初始化器中推斷，類似於隱式類型化的變數。

這個程式碼大部份都是在宣告 OracleParameter 參數，注意它們對應的資料庫類型型態。ExecuteNonQuery() 方法會返回命令所影響的筆數。

可以把 SQL 改成 Upsert，讓生活可以優雅一些:

src/ExecuteSample.cs

string sql = "MERGE INTO emp e " +
             "USING(select :empno as empno, " +
             "             :ename as ename, " +
             "             :job as job, " +
             "             :mgr as mgr, " +
             "             :hiredate as hiredate, " +
             "             :sal as sal, " +
             "             :comm as comm, " +
             "             :deptno as deptno " +
             "        from dual) p " +
             "ON(e.empno = p.empno) " +
             "WHEN MATCHED THEN " +
             "UPDATE SET e.ename = p.ename, " +
             "           e.job = p.job, " +
             "           e.mgr = p.mgr, " +
             "           e.hiredate = p.hiredate, " +
             "           e.sal = p.sal, " +
             "           e.comm = p.comm, " +
             "           e.deptno = p.deptno " +
             "WHEN NOT MATCHED THEN " +
             "  INSERT(e.empno, e.ename, e.job, e.mgr, e.hiredate, e.sal, e.comm, e.deptno) " +
             "  VALUES(p.empno, p.ename, p.job, p.mgr, p.hiredate, p.sal, p.comm, p.deptno)";

ExecuteScalar( ) 方法

在許多情況下，需要從 SQL 語句返回一個結果，例如要查詢資料表的筆數，或者資料庫伺服器當前的日期時間。ExecuteScalar()方法就可用於這些場合。同樣在 ExecuteSample.cs 中新增一個靜態方法 ExecuteScalar():

src/ExecuteSample.cs

public static void ExecuteScalar()
{
  string sql = "SELECT count(*) as count From emp";

  try
  {
    using (var connection = new OracleConnection(GetConnectionString()))
    using (var command = new OracleCommand(sql, connection))
    {
      connection.Open();
      object count = command.ExecuteScalar();
      Console.WriteLine($"counted {count} emp records");
    }
  }
  catch (OracleException ex)
  {
    Console.WriteLine(ex.Message);
  }
}

該方法返回一個物件，根據需要，可以把該物件強制轉換為合適的類型。如果所調用的 SQL 只返回一列(column)，則最好使用 ExecuteScalar( ) 方法來檢索這一列。這也適合用於只返回一個值的存儲過程(Stored Procedure)。

整合 RabbitMQ 消費訊息與 Oracle

終於來到最終的目標，將 RabbitMQ 的消費訊息寫入 Oracle 資料庫。

回到 Receive 專案目錄下，首先需要安裝 3 個 NuGet 套件:

cd Receive
dotnet add package Oracle.ManagedDataAccess.Core
dotnet add package Microsoft.Extensions.Configuration
dotnet add package Microsoft.Extensions.Configuration.Json

將 OracleSample 專案目錄下的 config.json 檔案複製到 Receive 專案目錄下。

在 Receive 專案目錄下新建一個子目錄 Services，我們要將一些後端資料庫的服務放在這個目錄下。在 Services 目錄下新建一個 C# 程式檔 OracleDemoMessage.cs:

Services/OracleDemoMessage.cs

using System;
using System.IO;
using Oracle.ManagedDataAccess.Client;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Configuration.Json;

namespace Receive.Services
{
  public class OraDemoMessage
  {
    public static void Save(string messageParameter, string createdbyParameter)
    {
      string sql = "INSERT INTO DEMO_MESSAGES (message, createdby) " +
                   "VALUES (:message, :createdby)";
      try
      {
        using (var connection = new OracleConnection(GetConnectionString()))
        using (var command = new OracleCommand(sql, connection))
        {
          command.BindByName = true;

          OracleParameter messageBind = new OracleParameter("message", OracleDbType.Varchar2);
          messageBind.Value = messageParameter;
          command.Parameters.Add(messageBind);

          OracleParameter createdbyBind = new OracleParameter("createdby", OracleDbType.Varchar2);
          createdbyBind.Value = createdbyParameter;
          command.Parameters.Add(createdbyBind);

          connection.Open();
          int rows = command.ExecuteNonQuery();
          Console.WriteLine($"rowsAffected: {rows} row(s)");
        }
      }
      catch (OracleException ex)
      {
        Console.WriteLine(ex.Message);
      }
    }

    private static string GetConnectionString()
    {
      var configurationBuilder = new ConfigurationBuilder()
        .SetBasePath(Directory.GetCurrentDirectory())
        .AddJsonFile("config.json");
      IConfiguration config = configurationBuilder.Build();
      string connectionString = config["Data:DefaultConnection:ConnectionString"];
      return connectionString;
    }
  }
}

這裡有一個接受兩個參數的 Save() 靜態方法，這會將資料寫入 Oracle 資料庫的 DEMO_MESSAGES 資料表。

回到 Receive 專案目錄，我們需要修改專案的入口點 Receive.cs 程式，首先要加入 Receive.Services 命名空間:

Receive.cs

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Receive.Services;

再來修改消費者事件監聽器，將消費消息寫入 Oracle 資料庫:

Receive.cs

consumer.Received += (model, e) =>
{
  var body = e.Body;
  var message = Encoding.UTF8.GetString(body);

  Console.WriteLine($" [x] Received {message}");

  OraDemoMessage.Save(message, "RabbitMQ dotnetDemoQueue");
};

現在就可以啟動接收器 Receive:

dotnet run

 Press [enter] to exit.

開啟另一個終端視窗，切換到 Send 專案目錄下送出一個訊息:

cd Send
dotnet run

 [x] Send 哈囉 from C# send 消息 2019/7/17 下午 03:53:06
Press [enter] to exit.

到 Oracle 資料庫驗證:

SQL

SQL> select * from demo_messages;

   ID MESSAGE                                          CREATED    CREATEDBY
----- ------------------------------------------------ ---------- ------------------------------

   23 哈囉 from C# send 消息 2019/7/17 下午 03:53:06     17-JUL-19  RabbitMQ dotnetDemoQueue

就這樣了，祝好運!