Аксон создает агрегат внутри саги
С меньшей серией, вам лучше просто сортировать, а затем взять голову / хвост!
Это запрос функции панды , должен видеть в 0.14 (нужно преодолеть некоторые биты с разными типами данных), эффективное решение для более крупных рядов (> 1000 элементов) использует kth_smallest из pandas algos (предупреждение этой функции мутирует массив, к которому он применяется, чтобы использовать копию!):
In [11]: s = pd.Series(np.random.randn(10))
In [12]: s
Out[12]:
0 0.785650
1 0.969103
2 -0.618300
3 -0.770337
4 1.532137
5 1.367863
6 -0.852839
7 0.967317
8 -0.603416
9 -0.889278
dtype: float64
In [13]: n = 3
In [14]: pd.algos.kth_smallest(s.values.astype(float), n - 1)
Out[14]: -0.7703374582084163
In [15]: s[s <= pd.algos.kth_smallest(s.values.astype(float), n - 1)]
Out[15]:
3 -0.770337
6 -0.852839
9 -0.889278
dtype: float64
Если вы хотите это по порядку:
In [16]: s[s <= pd.algos.kth_smallest(s.values.astype(float), n - 1)].order()
Out[16]:
9 -0.889278
6 -0.852839
3 -0.770337
dtype: float64
Если вы беспокоитесь о дубликатах (присоединитесь к n-му месту), вы можете взять голову:
In [17]: s[s <= pd.algos.kth_smallest(s.values.astype(float), n - 1)].order().head(n)
Out[17]:
9 -0.889278
6 -0.852839
3 -0.770337
dtype: float64
2 ответа
Решение, которое вы разместили, определенно подойдет. Позвольте мне объяснить сценарий, который вы здесь использовали, хотя и для других людей.
В приложении Axon Framework 4.x любой компонент обработки событий, а значит и ваши экземпляры Saga, поддерживаются TrackingEventProcessor. Процессор отслеживания событий «отслеживает», в какой точке потока событий он обрабатывает события. Он сохраняет эту информацию через TrackingToken, для которого TokenStore является делегирующей частью работы.
Однако, если вы не указали TokenStore, у вас будет в памяти TrackingTokens для каждого обработчика событий отслеживания. Это означает, что при перезапуске ваш процессор отслеживания событий думает: «О, я еще не занимался обработкой событий, позвольте мне начать с начала времени». Из-за этого ваши экземпляры Saga будут каждый раз начинать новый, пытаясь воссоздать данный экземпляр Aggregate.
Впредь, указав TokenStore, как вы решили проблему, которая у вас была.
Обратите внимание, что в среде Spring Boor, например, при наличии стартера Spring Data, Axon автоматически создаст для вас JpaTokenStore.
Я решил свою проблему, просто добавив конфигурацию хранилища токенов, она делает именно то, что мне нужно - отслеживать обработанные события.
Базовая конфигурация пружины:
@Bean
fun tokenStore(client: MongoClient): TokenStore = MongoTokenStore.builder()
.mongoTemplate(DefaultMongoTemplate.builder().mongoDatabase(client).build())
.serializer(JacksonSerializer.builder().build())
.build()