mirror of
https://github.com/go-admin-team/go-admin.git
synced 2026-09-22 10:33:13 +00:00
zhangwenjian
4510b06959
setupQueue ended with `go queueAdapter.Run()`, and the three log consumers were registered afterwards, from setup(). The contract implementations refuse a registration once the queue is running - memqueue and the redis queue both answer storage.ErrQueueAlreadyStarted - and Register cannot report it: it returns nothing, which its own comment in core records as the reason the interface is deprecated. Start first and register second, across two goroutines, and the registration is dropped without the caller being able to tell. What follows is not quiet. No consumer group was created, so redis refuses every later publish with storage.ErrNoHandler, and go-admin logs that at error level from both call sites while the login and operation log rows are simply never written. The silence is in the registration; the cost shows up on every request after it. Which implementation is behind the interface depends on the configuration. config.QueueConfig.Setup returns queue.NewMemory directly when there is no redis section - and that one does not care about the order, because its Register just starts another consumer goroutine. Only a redis section reaches storage.LegacyQueueAdapter, which wraps the contract implementation and therefore refuses. So the defect is invisible in the default deployment and shows up only where redis is configured, dropping the login log, the operation log and the api check - the three things #892 was about. The start therefore moves to the code that registers, and nothing starts the queue but that. The registration also moves onto AfterResource. It has to: a reload rebuilds the adapter, and consumers attached to the one that existed at start-up are attached to a queue nobody publishes to any more. Being on that phase means running again on every reload, so the callback is idempotent with respect to a given queue rather than "does nothing the second time" - registering twice on the same queue would give every message two consumers and write every log row twice. Identity for that comes from common/storage, where the adapter is built, as a generation counter. It cannot come from the accessors: GetQueueAdapter and GetQueuePrefix build a fresh runtime.Queue wrapper on every call, so comparing two of them compares two wrappers and never matches however many times the adapter underneath has been replaced. A counter also keeps the comparison on a uint64 rather than an `==` between two interface values, which would panic on an adapter type that is not comparable. Generation 0 means the configuration has no queue section, so nothing was installed and the runtime hands back its own memory queue. That case still gets consumers, because the registration this replaces was unconditional and dropping it would stop the logs for anyone who commented the section out. Two things fixed on the way past: - `if q := sdk.Runtime.GetQueueAdapter(); q != nil { q.Shutdown() }` was always true. GetQueueAdapter never returns nil - with nothing configured the runtime falls back to its own memory queue and wraps that - so the first start shut down the fallback queue before anything had used it. Only an adapter this package installed is shut down now. - config.Setup becomes bootstrap.SetupConfig, which is what announces AfterResource, and announces it after the callbacks that build the resources rather than before. attachConsumersOnce is split out so the order and the once-ness can be checked against a queue the test controls; neither can be read back out of a real adapter. Four tests cover the ordering, both directions of the idempotency rule, and the unconfigured case. Both counter-proofs compile and fail: calling Run before the registrations reports each of the three as "came after Run", and dropping the generation guard reports eight calls where four are wanted. One honest limit: the counter-proof for the ordering makes Run synchronous. The original arrangement started the queue on another goroutine, and a race cannot be made to fail every time - which is the reason the order is enforced by structure here instead of being left to be noticed in use.
100 lines
3.2 KiB
Go
100 lines
3.2 KiB
Go
/*
|
|
* @Author: zhangwenjian
|
|
* @Date: 2025/04/13 22:03
|
|
* @Last Modified by: zhangwenjian
|
|
* @Last Modified time: 2025/04/13 22:03
|
|
*/
|
|
|
|
package storage
|
|
|
|
import (
|
|
"log"
|
|
"sync"
|
|
|
|
"github.com/go-admin-team/go-admin-core/v2/captcha"
|
|
"github.com/go-admin-team/go-admin-core/v2/sdk"
|
|
"github.com/go-admin-team/go-admin-core/v2/sdk/config"
|
|
)
|
|
|
|
// Setup 配置storage组件
|
|
func Setup() {
|
|
setupCache()
|
|
setupCaptcha()
|
|
setupQueue()
|
|
}
|
|
|
|
func setupCache() {
|
|
cacheAdapter, err := config.CacheConfig.Setup()
|
|
if err != nil {
|
|
log.Fatalf("cache setup error, %s\n", err.Error())
|
|
}
|
|
sdk.Runtime.SetCacheAdapter(cacheAdapter)
|
|
}
|
|
|
|
func setupCaptcha() {
|
|
captcha.SetStore(captcha.NewCacheStore(sdk.Runtime.GetCacheAdapter(), 600))
|
|
}
|
|
|
|
var (
|
|
queueMu sync.Mutex
|
|
// installed is the adapter setupQueue built, kept so the next reload can
|
|
// shut it down, and counted so a consumer can tell one from the next.
|
|
installed interface{ Shutdown() }
|
|
installedGen uint64
|
|
)
|
|
|
|
// QueueGeneration reports how many times this package has installed a queue
|
|
// adapter. It changes every time setupQueue builds a new one, which is on
|
|
// every configuration reload, and stays 0 for as long as the configuration has
|
|
// no queue section at all - in which case nothing is installed and callers are
|
|
// working with the runtime's own fallback queue.
|
|
//
|
|
// It exists because there is no way to ask for the adapter's identity from the
|
|
// outside. sdk.Runtime.GetQueueAdapter and GetQueuePrefix build a fresh
|
|
// runtime.Queue wrapper on every call, so comparing what two calls return
|
|
// compares two wrappers and never matches, however many times the underlying
|
|
// adapter has been replaced. This package creates the adapter, so this is the
|
|
// only place that knows. A counter rather than the adapter itself keeps the
|
|
// comparison on a uint64: an adapter type that is not comparable would panic
|
|
// an `==` between two interface values.
|
|
func QueueGeneration() uint64 {
|
|
queueMu.Lock()
|
|
defer queueMu.Unlock()
|
|
return installedGen
|
|
}
|
|
|
|
func setupQueue() {
|
|
if config.QueueConfig.Empty() {
|
|
return
|
|
}
|
|
|
|
queueMu.Lock()
|
|
defer queueMu.Unlock()
|
|
|
|
// Only an adapter this package installed. GetQueueAdapter never returns
|
|
// nil - with nothing configured the runtime falls back to its own memory
|
|
// queue and wraps that - so the `if q := GetQueueAdapter(); q != nil` this
|
|
// replaces was always true, and shut down the fallback queue on the very
|
|
// first start, before anything had used it.
|
|
if installed != nil {
|
|
installed.Shutdown()
|
|
}
|
|
|
|
queueAdapter, err := config.QueueConfig.Setup()
|
|
if err != nil {
|
|
log.Fatalf("queue setup error, %s\n", err.Error())
|
|
}
|
|
sdk.Runtime.SetQueueAdapter(queueAdapter)
|
|
installed = queueAdapter
|
|
installedGen++
|
|
|
|
// Deliberately not started here. Run has to come after the consumers have
|
|
// registered: the contract implementations refuse a registration once the
|
|
// queue is running (storage.ErrQueueAlreadyStarted), and the legacy
|
|
// adapter this repository still goes through swallows that error rather
|
|
// than reporting it - its own comment says the interface gives it no way
|
|
// to tell the caller. Starting here and registering afterwards is
|
|
// therefore a race that loses consumers in silence. Whoever registers is
|
|
// the one that starts it.
|
|
}
|