是的，没有实现 worker，但我现在的用法是和 sidekiq 一起，利用 sidekiq 的 worker 来实现，可以看上面新提交的代码

if Sidekiq.server?
  Sidekiq.on(:startup) do
    BQ = BufferQueue.new(max_batch_size: 10, execution_interval: 100) do |batch|
      puts "bulk insert #{batch.size} records"
      Hyper::Event.import(
        EventJob::COLUMN_NAMES,
        batch.flatten.map { |attr| Hyper::Event.new(attr) },
        validate: false,
        timestamps: false
      ) unless batch.empty?
    end
  end
end

in sidekiq job:

class EventJob
  include Sidekiq::Worker

  COLUMN_NAMES = Hyper::Event.column_names

  def perform(*args)
      result = a_lot_of_cal
      BQ.push result
    end
  end
end

效果就是先利用 sidekiq 的多线程并行处理一些计算任务，产生的结果批量插入数据库； 相比之前 sidekiq job 里单条插入会快很多。并且可以利用 activerecord import 把多条记录改成 insert into values 的形式，插入更快。

ruby 的内置 queue 不太好用：https://bugs.ruby-lang.org/issues/9145

效果

[1] pry(main)> bq = BufferQueue.new(max_batch_size: 10, execution_interval: 5) do |batch|
  p batch
end
25.times do |i|
  bq.push(i)
end
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
=> 25
[2] pry(main)> [20, 21, 22, 23, 24]

[2] pry(main)> quit
shutdown ...

搞了一个简单版 https://github.com/HyperCable/hypercable/pull/49/files

# frozen_string_literal: true

class BufferQueue
  attr_reader :max_batch_size, :execution_interval, :timeout_interval, :callback
  def initialize(max_batch_size: 100, execution_interval: 60, timeout_interval: 60, &callback)
    @max_batch_size = max_batch_size
    @execution_interval = execution_interval
    @timeout_interval = timeout_interval
    @queue = Queue.new
    @timer = Concurrent::TimerTask.new(execution_interval: execution_interval, timeout_interval: timeout_interval) do
      flush
    end
    @timer.execute
    @callback = callback
    at_exit { shutdown }
  end

  def flush
    batch = []
    max_batch_size.times do
      if not @queue.empty?
        begin
          batch << @queue.pop(true)
        rescue ThreadError
          puts "queue is empty"
          break
        end
      else
        break
      end
    end
    callback.call(batch) unless batch.empty?
  end

  def push(item)
    @queue << item
    if @queue.size >= max_batch_size
      flush
    end
    item
  end

  def shutdown
    puts "shutdown ..."
    @timer.shutdown
    flush
  end
end

这种帖子底下需要一个 warning 功能

看起来不错

那个是用的 client middleware，要用的话，我要改成 server middleware。不一定基于 sidekiq，感觉 Ractor 可以做这东西。

相似度很高

可以把爬虫也替换了 https://github.com/pramsey/pgsql-http :

居然还有人在

还没 production ready 就被官方发现了

sdk 都是可以得到源码的，但经过混淆了。

也不需要开源，目前 hypercable analytics 的使用方式：

<script async src="https://www.googletagmanager.com/gtag/js?id=G-PJBVFLBFSB"></script>
<script>
  window.dataLayer = window.dataLayer || [];
  function gtag(){dataLayer.push(arguments);}
  gtag('js', new Date());
  gtag('config', 'G-PJBVFLBFSB', {transport_url: 'https://learnsql.io/c7f4edce-58c3-4917-8f18-a2ea6c1b93dc'});
</script>

新建的 ga 都是 google Analytic v4 的；

这里有介绍：https://support.google.com/analytics/answer/10089681

是 数据自有的需求很大 有了这些行为数据之后，做 crm 广告系统 推荐系统 也就很容易了

还有一个计划 就是 saas 版也可以直接暴露一个只读的数据库链接，直接使用或拉取原始数据到自己服务器。私有部署版当然就更灵活了

会的 该有的都可以有 只是目前需要做的太多了

目前好多 broken link，下个版本一起修。。

hoooopo@gmail.com 111111

还有很多更酷的特性等待发布

http://www.postgres.cn/docs/9.4/citext.html

。。。

admin 里

👌