[Ruby] Gem なしで API に並行でリクエストを送信する

## やりたいこと

ある API に複数のリクエストを送信するために以下のコードを用意した。

```rb
require 'net/http'

def post(url:, body:, content_type: 'application/json')
  uri = URI.parse(url)
  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = uri.scheme == 'https'
  request = Net::HTTP::Post.new(uri.path, 'Content-Type' => content_type)
  request.body = body
  http.request(request)
end

def post_names(names)
  names.map do |name|
    response = post(url: 'https://jsonplaceholder.typicode.com/posts', body: { name: name }.to_json)
    { tid: Thread.current.object_id, code: response.code, body: JSON.parse(response.body) }
  end
end
```
```rb
require 'benchmark'

post_names(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌))
#=> 
# [{tid: 334792, code: "201", body: {"name" => "🐕", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🐈", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🐇", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🐏", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🐬", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🐊", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🦆", "id" => 101}},
#  {tid: 334792, code: "201", body: {"name" => "🦌", "id" => 101}}]
Benchmark.realtime { post_names(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌)) }.round(2)
#=> 3.85
```

この API へのリクエスト送信を並行に行い、実行時間をもっと早くできないだろうか？

## 方法

Thread を使ってリクエストを並行に送信する。

```rb
def post_names_concurrently(names)
  threads = names.map do |name|
    Thread.new(name) do |name|
      response = post(url: 'https://jsonplaceholder.typicode.com/posts', body: { name: name }.to_json)
      { tid: Thread.current.object_id, code: response.code, body: JSON.parse(response.body) }
    end
  end

  threads.map(&:value)
end
```

この方法で実行時間を 3.85 秒から 0.57 秒に短縮できた。

```rb
post_names_concurrently(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌))
#=> 
# [{tid: 513672, code: "201", body: {"name" => "🐕", "id" => 101}},
#  {tid: 513688, code: "201", body: {"name" => "🐈", "id" => 101}},
#  {tid: 513656, code: "201", body: {"name" => "🐇", "id" => 101}},
#  {tid: 513640, code: "201", body: {"name" => "🐏", "id" => 101}},
#  {tid: 513632, code: "201", body: {"name" => "🐬", "id" => 101}},
#  {tid: 513648, code: "201", body: {"name" => "🐊", "id" => 101}},
#  {tid: 513664, code: "201", body: {"name" => "🦆", "id" => 101}},
#  {tid: 513680, code: "201", body: {"name" => "🦌", "id" => 101}}]
Benchmark.realtime { post_names_concurrently(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌)) }.round(2)
#=> 0.57
```

しかし、この方法では `names` の要素数の数だけ Thread を生成してしまう。仮に要素数が (1,000 以上など) 大量になると、それに比例してリソースを消費してしまう。そこで生成する Thread 数の上限を指定できるようにする。

```rb
def post_names_concurrently(names, thread_count: 5)
  # Thread 間で names を共有して操作するために相互排他ロック (Mutal Exclusion) を使う。
  mutex = Mutex.new
  
  threads = thread_count.times.map do
    Thread.new(names) do |names_in_thread|
      results = []

      loop do
        name = mutex.synchronize { names_in_thread.pop }
        break unless name
        response = post(url: 'https://jsonplaceholder.typicode.com/posts', body: { name: name }.to_json)
        results.push(tid: Thread.current.object_id, code: response.code, body: JSON.parse(response.body))
      end

      results
    end
  end

  threads.flat_map(&:value)
end
```

- Thread 数の上限を減らすと実行時間が増えてしまう
- 結果の並び順がランダムになる

というデメリットはあるものの、生成する Thread を任意の個数に絞り込むことができるようになった。

```rb
post_names_concurrently(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌), thread_count: 3)
#=> 
# [{tid: 100008, code: "201", body: {"name" => "🦌", "id" => 101}},
#  {tid: 100008, code: "201", body: {"name" => "🐇", "id" => 101}},
#  {tid: 100000, code: "201", body: {"name" => "🦆", "id" => 101}},
#  {tid: 100000, code: "201", body: {"name" => "🐏", "id" => 101}},
#  {tid: 100000, code: "201", body: {"name" => "🐕", "id" => 101}},
#  {tid: 99992, code: "201", body: {"name" => "🐊", "id" => 101}},
#  {tid: 99992, code: "201", body: {"name" => "🐬", "id" => 101}},
#  {tid: 99992, code: "201", body: {"name" => "🐈", "id" => 101}}]
Benchmark.realtime { post_names_concurrently(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌), thread_count: 3) }.round(2)
#=> 1.42

post_names_concurrently(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌), thread_count: 5)
#=>
# [{tid: 106344, code: "201", body: {"name" => "🦌", "id" => 101}},
#  {tid: 106344, code: "201", body: {"name" => "🐈", "id" => 101}},
#  {tid: 106336, code: "201", body: {"name" => "🦆", "id" => 101}},
#  {tid: 106336, code: "201", body: {"name" => "🐇", "id" => 101}},
#  {tid: 106352, code: "201", body: {"name" => "🐊", "id" => 101}},
#  {tid: 106352, code: "201", body: {"name" => "🐕", "id" => 101}},
#  {tid: 106368, code: "201", body: {"name" => "🐬", "id" => 101}},
#  {tid: 106360, code: "201", body: {"name" => "🐏", "id" => 101}}]
Benchmark.realtime { post_names_concurrently(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌), thread_count: 5) }.round(2)
#=> 0.79
```

| Thead 数 | 実行時間 (秒) | 備考 |
|--|--|--|
| 1 | 3.85 | 直列 |
| 3 | 1.42 | |
| 5 | 0.79 | |
| 8 | 0.57 | 要素数 = Thread 数 |

## バージョン情報

```shell-session
$ ruby -v
ruby 3.4.1 (2024-12-25 revision 48d4efcb85) +PRISM [arm64-darwin24]
```

## おまけ: Typhoeus という Gem を使う

Gem を使う場合は [Typhoeus](https://github.com/typhoeus/typhoeus) の Typhoeus::Hydra を使うとよい。

```shell-session
$ gem install typhoeus
Fetching typhoeus-1.4.1.gem
Fetching ffi-1.17.1-arm64-darwin.gem
Fetching ethon-0.16.0.gem
Successfully installed ffi-1.17.1-arm64-darwin
Successfully installed ethon-0.16.0
Successfully installed typhoeus-1.4.1
3 gems installed
```

```rb
require 'typhoeus'

def post_names_concurrently_with_typhoeus(names, max_concurrency:)
  hydra = Typhoeus::Hydra.new(max_concurrency: max_concurrency)
  results = []

  names.each do |name|
    request = Typhoeus::Request.new(
      'https://jsonplaceholder.typicode.com/posts',
      method: :post,
      body: { name: name }.to_json,
      headers: { 'Content-Type' => 'application/json' }
    )

    request.on_complete do |response|
      results << { tid: Thread.current.object_id, code: response.response_code, body: JSON.parse(response.body) }
    end

    hydra.queue(request)
  end

  hydra.run

  results
end
```

```rb
require 'benchmark'

post_names_concurrently_with_typhoeus(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌), max_concurrency: 5)
#=> 
# [{tid: 21696, code: 201, body: {"name" => "🐕", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🐈", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🐇", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🐏", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🐬", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🐊", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🦆", "id" => 101}},
#  {tid: 21696, code: 201, body: {"name" => "🦌", "id" => 101}}]
Benchmark.realtime { post_names_concurrently_with_typhoeus(%w(🐕 🐈 🐇 🐏 🐬 🐊 🦆 🦌), max_concurrency: 5) }.round(2)
#=> 1.27
```

なお、Typhoeus::Hydra は Thread ではなく libcurl の multi-handle という仕組みを使って並行処理を行っているらしい (詳細は未調査) 。

## 参考

- [asynchronous - Parallel HTTP requests in ruby - Stack Overflow](https://stackoverflow.com/questions/8778732/parallel-http-requests-in-ruby)
- [library net/http (Ruby 3.4 リファレンスマニュアル)](https://docs.ruby-lang.org/ja/latest/library/net=2fhttp.html)
- [class Thread (Ruby 3.4 リファレンスマニュアル)](https://docs.ruby-lang.org/ja/latest/class/Thread.html)
- [class Thread::Mutex (Ruby 3.4 リファレンスマニュアル)](https://docs.ruby-lang.org/ja/latest/class/Thread=3a=3aMutex.html#I_SYNCHRONIZE)