Tengo un modelo llamado AisSignal con aproximadamente 3000 registros y estoy comparando cada uno con otro modelo llamado Footprint con aproximadamente 10 registros, por lo que tenemos un bucle de 3000 x 10.
Lo intenté:
Parallel.each(AisSignal.all, in_processes: 8) do |signal| Footprint.all.each do |footprint| if footprint.cover([signal.lon, signal.lat]) signal.update(imo: 'in') break end end endpero se ejecuta en 10 segundos como un bloque normal.
Traté de cambiar de procesos a subprocesos como se muestra a continuación, pero esto hace que la aplicación se congele.
Parallel.each(AisSignal.all, in_threads: 8) do |signal| Footprint.all.each do |footprint| if footprint.cover([signal.lon, signal.lat]) signal.update(imo: 'in') break end end endTengo 50 tamaños de grupo en base de datos.yml
Cualquier idea o enfoque para tener varios subprocesos que se ejecutan en paralelo para actualizar registros. Tendré que actualizar más registros, lo que puede llevar unos minutos.
Los hilos y las bifurcaciones a menudo no funcionan bien con las conexiones de la base de datos. Si no se maneja correctamente, los subprocesos/procesos pueden terminar intentando usar la misma conexión al mismo tiempo.
Parallel menciona esto en su documentación . Necesita hacer uso de la agrupación de conexiones .
Un grupo de conexiones sincroniza el acceso de subprocesos a un número limitado de conexiones de base de datos. La idea básica es que cada subproceso verifica una conexión de base de datos del grupo, usa esa conexión y vuelve a verificar la conexión. ConnectionPool es completamente seguro para subprocesos y garantizará que dos subprocesos no puedan usar una conexión al mismo tiempo. , siempre que se cumpla correctamente el contrato de ConnectionPool. También manejará casos en los que hay más subprocesos que conexiones: si todas las conexiones se han desprotegido y un subproceso intenta desproteger una conexión de todos modos, ConnectionPool esperará hasta que otro subproceso haya verificado una conexión.
Parallel.each(AisSignal.all, in_threads: 8) do |signal| ActiveRecord::Base.connection_pool.with_connection do Footprint.all.each do |footprint| if footprint.cover([signal.lon, signal.lat]) signal.update(imo: 'in') break end end end endTenga en cuenta que este código es muy ineficiente.
AisSignal .Footprint . Utilizará mucha memoria y se ejecutará en tiempo s*f, donde s es el número de señales y f es el número de huellas.
Puede reducir el consumo de memoria reemplazando Footprint.all.each con Footprint.find_each . Esto cargará filas en lotes.
La creación de subprocesos no es la forma de hacer que las consultas a la base de datos sean más rápidas . El problema fundamental es que está escaneando Footprint varias veces en Ruby en lugar de dejar que la base de datos lo haga. if footprint.cover([signal.lon, signal.lat]) debería ser una cláusula where.
AisSignal.find_each do |signal| # With ... being the equivalent of `cover([signal.lon, signal.lat])` # as a where clause. signal.update!(imo: 'in') if Footprint.exists?(...) endEsto podría hacerse aún más rápido como una unión.
# ... is the equivalent of `cover([signal.lon, signal.lat])` AisSignal.joins("inner join footprints on ...").update_all(imo: 'in')