蒋先森のBlog

/**
 * 模拟流式结果输出
 */
@GetMapping(value = "stream_query", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
public SseEmitter streamQuery() {
    SseEmitter emitter = new SseEmitter(60000L);
    Flux.interval(Duration.ofSeconds(1))
            .map(data -> "Server Message # " + data)
            .doOnCancel(emitter::complete)
            .subscribe(data -> {
                try {
                    emitter.send(SseEmitter.event().data(data));
                } catch (Exception e) {
                    log.error("发送数据异常", e);
                }
            });
    return emitter;
}

public static void main(String[] args) {
    JSONObject data = new JSONObject();
    data.put("a", "a");
    data.put("b", "b");
    JSONObject after = new JSONObject(data);
    after.put("c", "c");
    System.out.println("原始数据：" + data);
    System.out.println("修改后：" + after);
}

private static final int DEFAULT_INITIAL_CAPACITY = 16;
private final Map<String, Object> map;

public JSONObject(Map<String, Object> map) {
    if (map == null) {
        throw new IllegalArgumentException("map is null.");
    } else {
        this.map = map;
    }
}

public List<UserInfo> getListByName(String name) {
  LambdaQueryWrapper<UserInfo> queryWrapper = new LambdaQueryWrapper<>();
  queryWrapper.eq(UserInfo::getName, name);
  return list(queryWrapper);
}

@FunctionalInterface
public interface Func<E, R> extends Function<E, R>, Serializable {

}

public class ReflectLambdaUtils {

    private static final Logger logger = LoggerFactory.getLogger(ReflectLambdaUtils.class);

    /**
     * 根据参数获取字段名称
     *
     * @param func
     * @return
     */
    public static <E, R> String getFieldName(Func<E, R> func) {
        Field field = getField(func);
        return field.getName();
    }

    /**
     * 获取表达式的字段
     *
     * @param func
     * @return
     */
    public static <E, R> Field getField(Func<E, R> func) {
        try {
            Method method = func.getClass().getDeclaredMethod("writeReplace");
            method.setAccessible(Boolean.TRUE);
            // 1.调用 writeReplace 方法，返回一个 writeReplace 对象
            SerializedLambda serializedLambda = (SerializedLambda) method.invoke(func);
            String getterMethod = serializedLambda.getImplMethodName();
            String fieldName = Introspector.decapitalize(getterMethod.replace("get", ""));
            // 2. 获取的Class是字符串，并且包名是“/”分割，需要替换成“.”，才能获取到对应的Class对象
            String declaredClass = serializedLambda.getImplClass().replace("/", ".");
            Class<?> aClass = Class.forName(declaredClass, false, ClassUtils.getDefaultClassLoader());
            // 3.通过Spring 中的反射工具类获取Class中定义的Field
            return ReflectionUtils.findField(aClass, fieldName);
        } catch (ReflectiveOperationException e) {
            logger.error("解析类字段出现异常", e);
            throw new MongoPlusException("解析字段时出现异常");
        }
    }
}

@Slf4j
@Component
public class SampleJob {

    @Scheduled(cron = "* * * * * ?")
    public void printLog() {
        long time = System.currentTimeMillis();
        log.info("当前时间：{}", time);
    }
}

@Slf4j
@Component
public class DynamicCronJob implements SchedulingConfigurer {

    /**
     * 默认定时任务执行时间
     */
    private String taskCron = GlobalConstants.TASK_DEFAULT_CRON;

    @Override
    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        scheduledTaskRegistrar.addTriggerTask(() -> {
            long time = System.currentTimeMillis();
            log.info("执行任务，当前时间：{}", time);
        }, triggerContext -> {
            // 刷新cron时间
            CronTrigger cronTrigger = new CronTrigger(taskCron);
            Date nextExecuteTime = cronTrigger.nextExecutionTime(triggerContext);
            return nextExecuteTime;
        });
    }
        
    /**
     * 修改默认的定时任务时间
     * */
    public void setTaskCron(String taskCron) {
        this.taskCron = taskCron;
    }
}

// 标记线程池昨天，默认RUNNING
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 线程个数
private static final int COUNT_BITS = Integer.SIZE - 3;
// 线程最大个数
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// 线程池状态
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    // 获取当前线程池的状态+线程个数遍历组合值
    int c = ctl.get();
    // 当前线程池小于核心线程池，开启新线程执行
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 如果线程池处于RUNNING，将任务添加到队列中
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果队列满了，创建非核心线程执行任务，如果失败，执行拒绝策略
    else if (!addWorker(command, false))
        reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 检测队列是否为空
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
        // 循环CAS 增加线程数
        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 调用 w.run，实际调用 runWorker方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

@Override
public String[] selectImports(AnnotationMetadata annotationMetadata) {
	if (!isEnabled(annotationMetadata)) {
		return NO_IMPORTS;
	}
	AutoConfigurationMetadata autoConfigurationMetadata = AutoConfigurationMetadataLoader
			.loadMetadata(this.beanClassLoader);
	AnnotationAttributes attributes = getAttributes(annotationMetadata);
       // 所以的配置存放在configuration中，获取所有依赖的配置
	List<String> configurations = getCandidateConfigurations(annotationMetadata,
			attributes);
	configurations = removeDuplicates(configurations);
	Set<String> exclusions = getExclusions(annotationMetadata, attributes);
	checkExcludedClasses(configurations, exclusions);
	configurations.removeAll(exclusions);
	configurations = filter(configurations, autoConfigurationMetadata);
	fireAutoConfigurationImportEvents(configurations, exclusions);
	return StringUtils.toStringArray(configurations);
}