echart react 事件
2017-11-16 18:27:00 weixin_33724570 阅读数 31

Event 对SaltStack里面的每一个事件进行了记录,比job更加底层
Event可以记录Minion连接Master、Key认证、job等

Reactor 是基于每个Event事件做相应的操作。
Reactor监听Event,触发一些States操作

1. 查看Event事件

salt-run state.eveyt

[root@master ~]# salt-run state.event pretty=True
# 执行后在另外终端运行job salt '*' cmd.run 'hostname'
20171116165940375717    {
    "_stamp": "2017-11-16T08:59:40.376454", 
    "minions": [
        "salt-minion-01", 
        "salt-minion-02"
    ]
}
salt/job/20171116165940375717/new       {
    "_stamp": "2017-11-16T08:59:40.376951", 
    "arg": [
        "hostname"
    ], 
    "fun": "cmd.run", 
    "jid": "20171116165940375717", 
    "minions": [
        "salt-minion-01", 
        "salt-minion-02"
    ], 
    "tgt": "*", 
    "tgt_type": "glob", 
    "user": "root"
}
salt/job/20171116165940375717/ret/salt-minion-01        {
    "_stamp": "2017-11-16T08:59:40.416043", 
    "cmd": "_return", 
    "fun": "cmd.run", 
    "fun_args": [
        "hostname"
    ], 
    "id": "salt-minion-01", 
    "jid": "20171116165940375717", 
    "retcode": 0, 
    "return": "salt-minion-01", 
    "success": true
}
salt/job/20171116165940375717/ret/salt-minion-02        {
    "_stamp": "2017-11-16T08:59:40.419274", 
    "cmd": "_return", 
    "fun": "cmd.run", 
    "fun_args": [
        "hostname"
    ], 
    "id": "salt-minion-02", 
    "jid": "20171116165940375717", 
    "retcode": 0, 
    "return": "salt-minion-02", 
    "success": true
}
2. Reactor
示例: 接受Minion key,并完成初始化。
  • 首先看Minion key认证和认证完成后的Event:
    认证:
salt/auth       {
    "_stamp": "2017-11-16T09:22:04.444574", 
    "act": "pend", 
    "id": "master", 
    "pub": "-----BEGIN PUBLIC KEY-----\nMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAhcrNw/7KQzr6wFGKT42q\nc04m/w09tov
EQvyl/yJJ0XMrDdPiK7E8So9gIm/ykSnEuCmONTenpTA3A1qWo0oR\nHS6f4ucP4RuogxlCDIgip3+KXJMJi2vnpAm+zOIXi2ExxuxKa3f8sLLizaNXg2
/G\nVnB4I4CNqR/c3O6Qa3Q16zg8j1yeMDaRe1BwIJ2qv6Xno95Rdje1/bYo5m01XaQk\nuZuecOLdNScQLHsht57CJXRSuB8+S+ZPItg5p/jLRcB962v
H7L1zotJr4jFTSJS1\nthH92ymyFDFB72j4H+xNwvkv1niAnjEJ9Hxznt/Ra5FkdKdWh9FQEMqUJpC2DUlZ\ngwIDAQAB\n-----END PUBLIC KEY---
--", 
    "result": true
}

认证完成:

salt/minion/master/start        {
    "_stamp": "2017-11-16T09:54:19.513882", 
    "cmd": "_minion_event", 
    "data": "Minion master started at Thu Nov 16 17:54:19 2017", 
    "id": "master", 
    "pretag": null, 
    "tag": "salt/minion/master/start"
}
1 )配置reactor(在Master配置文件中)
reactor: 
  - 'salt/auth':
    - /srv/reactor/Minion.sls
  - 'salt/minion/master/start':
    - /srv/reactor/auto.sls

解释: 监控到 salt/auth行为时,执行/srv/reactor/Minion.sls操作;
监控到salt/minion/master/start行为时,执行/srv/reactor/auto.sls操作。

2)Minion.sls 文件
[root@master ~]# cat /srv/reactor/Minion.sls 
{% if 'act' in data and data['act'] == 'pend' and data['id'] == 'master' %}
key_accept:
  wheel.key.accept:
    - match: {{ data['id'] }}

{% endif %}

判断如果 act在 event事件中,且{'act':'pend'}{'id': 'master'},执行keep_accept操作,接受对应 data['id']。
接受master开头的Minion

3)auto.sls 文件
[root@master ~]# cat /srv/reactor/auto.sls 
run_state:
  local.state.sls:
    - tgt: {{ data['id'] }}
    - arg:
      - shencan

run_init:
  local.cmd.run:
    - tgt: {{ data['id'] }}
    - arg: 
      - echo initisok >> /tmp/cpis

发现 salt/minion/master/start 事件后,执行auto.sls文件;
针对目标data['id'] 执行shencan.sls 文件(file_root 中的state文件);
echo "initisok" 到Minion的/tmp/cpis 文件中

4)shencan.sls
[root@master ~]# cat /srv/salt/shencan.sls 
/tmp/example:
  file.managed:
    - source: salt://example

复制文件到Minion的/tmp文件夹下

转载于:https://www.jianshu.com/p/52ee266803f0

echart react 事件 相关内容

2016-06-10 02:15:00 weixin_34297704 阅读数 2

https://github.com/reactor/reactor/wiki/Getting-Started

 

http://www.codeweblog.com/spring%E7%9A%84reactor-%E7%9A%84%E5%BC%82%E6%AD%A5%E4%BA%8B%E4%BB%B6%E9%A9%B1%E5%8A%A8%E7%9A%84%E4%BB%8B%E7%BB%8D/

 

这个$()是导入的静态方法 

转载于:https://www.cnblogs.com/donaldlee2008/p/5573066.html

echart react 事件 相关内容

2016-05-10 06:12:00 weixin_34194317 阅读数 1

https://github.com/reactor

https://github.com/reactor/reactor-samples/

 

https://github.com/ReactiveX/RxJava

https://github.com/ReactiveX/RxJava/wiki/Getting-Started

http://reactivex.io/

https://github.com/ReactiveX/RxJava/wiki

转载于:https://www.cnblogs.com/donaldlee2008/p/5476420.html

echart react 事件 相关内容

2017-04-15 22:45:00 weixin_34242509 阅读数 190

maven

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bus</artifactId>
        </dependency>

config

@Configuration
public class EventConfig {

    @Bean
    Environment environment() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    @Autowired
    public EventBus eventBus(Environment environment, MyEventListener myEventListener) {
        EventBus eventBus = EventBus.create(environment, Environment.THREAD_POOL);

        eventBus.on($("myevent"), myEventListener);

        return eventBus;
    }
}

env的默认配置从reactor-core-2.0.8.RELEASE.jar!/META-INF/reactor/reactor-environment.properties中读取


#
# Copyright (c) 2011-2015 Pivotal Software Inc., Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

##
# Dispatcher configuration
#
# Each dispatcher must be configured with a type:
#
# reactor.dispatchers.<name>.type = <type>
#
# Legal values for <type> are dispatcherGroup, ringBuffer, synchronous, and workQueue.

# Depending on the type, further configuration is be possible:
#
# reactor.dispatchers.<name>.size:    dispatcherGroup and workQueue Dispatchers
# reactor.dispatchers.<name>.backlog: dispatcherGroup, ringBuffer, and workQueue Dispatchers
#
# A size less than 1 may be specified to indicate that the size should be the same as the number
# of CPUs.

# A thread pool executor dispatcher, named threadPoolExecutor
reactor.dispatchers.threadPoolExecutor.type = threadPoolExecutor
reactor.dispatchers.threadPoolExecutor.size = 0
# Backlog is how many Task objects to warm up internally
reactor.dispatchers.threadPoolExecutor.backlog = 2048

# A group of dispatchers replicated from the default dispatcher, named dispatcherGroup
reactor.dispatchers.dispatcherGroup.type = dispatcherGroup
reactor.dispatchers.dispatcherGroup.size = 0
reactor.dispatchers.dispatcherGroup.backlog = 2048

# A ring buffer dispatcher, named ringBuffer
reactor.dispatchers.shared.type = ringBuffer
reactor.dispatchers.shared.backlog = 8192

# A work queue dispatcher, named workQueue
reactor.dispatchers.workQueue.type = workQueue
reactor.dispatchers.workQueue.size = 0
reactor.dispatchers.workQueue.backlog = 2048

# The dispatcher named shared should be the default dispatcher
reactor.dispatchers.default = shared

event listener

@Component
public class MyEventListener implements Consumer<Event<MyEvent>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyEventListener.class);

    @Override
    public void accept(Event<MyEvent> eventContextEvent) {

        MyEvent event = eventContextEvent.getData();

        LOGGER.info("thread {} ,receive event:{}",Thread.currentThread().getName(),event.getData());
    }
}

publish event

    @Autowired
    EventBus eventBus;

    public void publishEvent(String data){
        eventBus.notify("myevent", Event.wrap(new MyEvent(data)));
    }

ThreadPoolExecutorDispatcher

reactor-core-2.0.8.RELEASE-sources.jar!/reactor/core/dispatch/ThreadPoolExecutorDispatcher.java
在reactor-core-2.0.8.RELEASE-sources.jar!/reactor/Environment.java创建默认的ThreadPoolExecutorDispatcher

private static ThreadPoolExecutorDispatcher createThreadPoolExecutorDispatcher(DispatcherConfiguration dispatcherConfiguration) {
        int size = getSize(dispatcherConfiguration, 0);
        int backlog = getBacklog(dispatcherConfiguration, 128);

        return new ThreadPoolExecutorDispatcher(size, backlog, dispatcherConfiguration.getName());
    }

构造器


/**
     * Create a new {@literal ThreadPoolExecutorDispatcher} with the given size, backlog, name, and {@link
     * java.util.concurrent.RejectedExecutionHandler}.
     *
     * @param poolSize
     *      the pool size
     * @param backlog
     *      the backlog size
     * @param threadName
     *      the name prefix to use when creating threads
     */
    public ThreadPoolExecutorDispatcher(int poolSize,
                                        int backlog,
                                        String threadName) {
        this(poolSize,
             backlog,
             threadName,
             new LinkedBlockingQueue<Runnable>(backlog),
             new RejectedExecutionHandler() {
                 @Override
                 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                     r.run();
                 }
             });
    }

默认采用的是LinkedBlockingQueue,大小为配置文件指定的backlog,RejectedExecutionHandler采用的是调用者执行。


/**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the {@code execute} method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

因此,队列没有满的时候是异步的,满的时候就阻塞了。

doc

echart react 事件 相关内容

2017-04-15 21:57:24 weixin_33910434 阅读数 27

maven

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bus</artifactId>
        </dependency>

config

@Configuration
public class EventConfig {

    @Bean
    Environment environment() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    @Autowired
    public EventBus eventBus(Environment environment, MyEventListener myEventListener) {
        EventBus eventBus = EventBus.create(environment, Environment.THREAD_POOL);

        eventBus.on($("myevent"), myEventListener);

        return eventBus;
    }
}

env的默认配置从reactor-core-2.0.8.RELEASE.jar!/META-INF/reactor/reactor-environment.properties中读取

#
# Copyright (c) 2011-2015 Pivotal Software Inc., Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

##
# Dispatcher configuration
#
# Each dispatcher must be configured with a type:
#
# reactor.dispatchers.<name>.type = <type>
#
# Legal values for <type> are dispatcherGroup, ringBuffer, synchronous, and workQueue.

# Depending on the type, further configuration is be possible:
#
# reactor.dispatchers.<name>.size:    dispatcherGroup and workQueue Dispatchers
# reactor.dispatchers.<name>.backlog: dispatcherGroup, ringBuffer, and workQueue Dispatchers
#
# A size less than 1 may be specified to indicate that the size should be the same as the number
# of CPUs.

# A thread pool executor dispatcher, named threadPoolExecutor
reactor.dispatchers.threadPoolExecutor.type = threadPoolExecutor
reactor.dispatchers.threadPoolExecutor.size = 0
# Backlog is how many Task objects to warm up internally
reactor.dispatchers.threadPoolExecutor.backlog = 2048

# A group of dispatchers replicated from the default dispatcher, named dispatcherGroup
reactor.dispatchers.dispatcherGroup.type = dispatcherGroup
reactor.dispatchers.dispatcherGroup.size = 0
reactor.dispatchers.dispatcherGroup.backlog = 2048

# A ring buffer dispatcher, named ringBuffer
reactor.dispatchers.shared.type = ringBuffer
reactor.dispatchers.shared.backlog = 8192

# A work queue dispatcher, named workQueue
reactor.dispatchers.workQueue.type = workQueue
reactor.dispatchers.workQueue.size = 0
reactor.dispatchers.workQueue.backlog = 2048

# The dispatcher named shared should be the default dispatcher
reactor.dispatchers.default = shared

event listener

@Component
public class MyEventListener implements Consumer<Event<MyEvent>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyEventListener.class);

    @Override
    public void accept(Event<MyEvent> eventContextEvent) {

        MyEvent event = eventContextEvent.getData();

        LOGGER.info("thread {} ,receive event:{}",Thread.currentThread().getName(),event.getData());
    }
}

publish event

    @Autowired
    EventBus eventBus;

    public void publishEvent(String data){
        eventBus.notify("myevent", Event.wrap(new MyEvent(data)));
    }

ThreadPoolExecutorDispatcher

reactor-core-2.0.8.RELEASE-sources.jar!/reactor/core/dispatch/ThreadPoolExecutorDispatcher.java
在reactor-core-2.0.8.RELEASE-sources.jar!/reactor/Environment.java创建默认的ThreadPoolExecutorDispatcher

private static ThreadPoolExecutorDispatcher createThreadPoolExecutorDispatcher(DispatcherConfiguration dispatcherConfiguration) {
        int size = getSize(dispatcherConfiguration, 0);
        int backlog = getBacklog(dispatcherConfiguration, 128);

        return new ThreadPoolExecutorDispatcher(size, backlog, dispatcherConfiguration.getName());
    }

构造器

/**
     * Create a new {@literal ThreadPoolExecutorDispatcher} with the given size, backlog, name, and {@link
     * java.util.concurrent.RejectedExecutionHandler}.
     *
     * @param poolSize
     *         the pool size
     * @param backlog
     *         the backlog size
     * @param threadName
     *         the name prefix to use when creating threads
     */
    public ThreadPoolExecutorDispatcher(int poolSize,
                                        int backlog,
                                        String threadName) {
        this(poolSize,
             backlog,
             threadName,
             new LinkedBlockingQueue<Runnable>(backlog),
             new RejectedExecutionHandler() {
                 @Override
                 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                     r.run();
                 }
             });
    }

默认采用的是LinkedBlockingQueue,大小为配置文件指定的backlog,RejectedExecutionHandler采用的是调用者执行。

/**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the {@code execute} method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

因此,队列没有满的时候是异步的,满的时候就阻塞了。

doc

echart react 事件 相关内容

为什么80%的码农都做不了架构师?>>>...

博文 来自: weixin_34417814
没有更多推荐了,返回首页