redux-saga原始碼解析
摘要：
Redux-saga是redux應用的又一個副作用模型。可以用來替換redux-thunk中介軟體。
redux-saga 抽象出Effect
（影響, 例如等待action、發出action、fetch資料等等），便於組合與測試。
我想在分析redux-saga之前，先來...
Redux-saga是redux應用的又一個副作用模型。可以用來替換redux-thunk中介軟體。 redux-saga 抽象出Effect （影響, 例如等待action、發出action、fetch資料等等），便於組合與測試。
我想在分析redux-saga之前，先來看看redux-thunk是怎麼一回事 redux 在我之前一篇文章中講過了連結 那我們就先用redux-thunk 來寫一個 asyncTodo 的demo
redux-thunk 分析
import { createStore, applyMiddleware } from 'redux';
const thunk = ({ dispatch, getState }) => next => action => {
if (typeof action === 'function') {
return action(dispatch, getState);
}
return next(action);
}
const logger = ({ getState }) => next => action => {
console.log('will dispatch', getState());
next(action)
console.log('state after dispatch', getState());
}
const todos = (state = [], action) => {
switch (action.type) {
case 'ADD_TODO':
return [
...state,
action.text
];
default:
return state
}
}
const store = createStore(
todos,
['Use Redux'],
applyMiddleware(logger, thunk),
);
store.dispatch(dispatch => {
setTimeout(() => {
dispatch({ type: 'ADD_TODO', text: 'Read the docs' });
}, 1000);
});
複製程式碼
原本redux中action只能是plain object ，redux-thunk使action可以為function。當我們想丟擲一個非同步的action時，其實我們是把非同步的處理放在了actionCreator中。 這樣就會導致action形式不統一，並且對於非同步的處理將會分散到各個action中，不利於維護。 接下來看看redux-saga是如何實現的
redux-saga
import { createStore, applyMiddleware } from 'redux';
import createSagaMiddleware from 'redux-saga';
import { put, take, fork, delay } from './redux-saga/effects'
import { delay as delayUtil } from 'redux-saga/utils';
// 獲取redux中介軟體
const sagaMiddleware = createSagaMiddleware({
sagaMonitor: {
// 列印 effect 便於分析redux-saga行為
effectTriggered(options) {
console.log(options);
}
}
})
function* rootSaga() {
const action = yield take('ADD_TODO_SAGA');
// delay(): { type: 'call', payload: { args: [1000], fn }}
yield delay(1000); // or yield call(delayUtil, 1000)
// put(): { type: 'PUT', payload: { action: {}, channel: null }}
yield put({ type: 'ADD_TODO', text: action.text});
}
const store = createStore(
todos,
['Use Redux'],
applyMiddleware(logger, sagaMiddleware),
);
// 啟動saga
sagaMiddleware.run(rootSaga);
store.dispatch({ type: 'ADD_TODO_SAGA', text: 'Use Redux-saga' });
複製程式碼
可以看到這裡丟擲的就是一個純action， saga在啟動之後監聽ADD_TODO_SAGA 事件，若事件發生執行後續程式碼。
原始碼
stdChannel
在開始createSagaMiddleware之前，先來了解一下channel redux-saga 通過 channel 接收與發出action與外部進行資料交換 在redux-saga中有三種 channel，分別是channel、eventChannel、multicastChannel； 在此我們僅僅分析一下用的最多的 multicastChannel
export function multicastChannel() {
let closed = false
// 這裡taker分為的currentTakers、 nextTakers的原因和redux subscribe類似，防止在遍歷taker時，taker發生變化。
let currentTakers = []
let nextTakers = currentTakers
const ensureCanMutateNextTakers = () => {
if (nextTakers !== currentTakers) {
return
}
nextTakers = currentTakers.slice()
}
const close = () => {
closed = true
const takers = (currentTakers = nextTakers)
for (let i = 0; i < takers.length; i++) {
const taker = takers[i]
taker(END)
}
nextTakers = []
}
return {
[MULTICAST]: true,
put(input) {
if (closed) {
return
}
if (isEnd(input)) {
close()
return
}
const takers = (currentTakers = nextTakers)
// 遍歷takers，找到與input匹配的taker並執行它。
for (let i = 0; i < takers.length; i++) {
const taker = takers[i]
if (taker[MATCH](input)) {
taker.cancel()
taker(input)
}
}
},
// 存下callback，與配置函式
take(cb, matcher = matchers.wildcard) {
if (closed) {
cb(END)
return
}
cb[MATCH] = matcher
ensureCanMutateNextTakers()
nextTakers.push(cb)
cb.cancel = once(() => {
ensureCanMutateNextTakers()
remove(nextTakers, cb)
})
},
close,
}
}
export function stdChannel() {
const chan = multicastChannel()
const { put } = chan
chan.put = input => {
if (input[SAGA_ACTION]) {
put(input)
return
}
// 暫時不用管
asap(() => put(input))
}
return chan
}
複製程式碼
createSagaMiddleware
獲取redux-middleware， 同時初始化runsaga函式，為後面啟動saga bind 所需的引數
export default function sagaMiddlewareFactory({ context = {}, ...options } = {}) {
const { sagaMonitor, logger, onError, effectMiddlewares } = options
let boundRunSaga
// redux middleware
function sagaMiddleware({ getState, dispatch }) {
// 新建一個channel
const channel = stdChannel()
channel.put = (options.emitter || identity)(channel.put)
boundRunSaga = runSaga.bind(null, {
context,
channel,
dispatch,
getState,
sagaMonitor,
logger,
onError,
effectMiddlewares,
})
return next => action => {
if (sagaMonitor && sagaMonitor.actionDispatched) {
sagaMonitor.actionDispatched(action)
}
const result = next(action) // hit reducers
// 將事件傳遞給saga
channel.put(action)
return result
}
}
// 啟動saga
sagaMiddleware.run = (...args) => {
// ...
return boundRunSaga(...args)
}
//...
return sagaMiddleware
}
複製程式碼
runsaga
export function runSaga(options, saga, ...args) {
// generate iterator
const iterator = saga(...args)
const {
channel = stdChannel(),
dispatch,
getState,
context = {},
sagaMonitor,
logger,
effectMiddlewares,
onError,
} = options
const effectId = nextSagaId()
// 一些錯誤檢查
// ...
const log = logger || _log
const logError = err => {
log('error', err)
if (err && err.sagaStack) {
log('error', err.sagaStack)
}
}
const middleware = effectMiddlewares && compose(...effectMiddlewares)
// 可以先理解為 finalizeRunEffect = runEffect => runEffect
const finalizeRunEffect = runEffect => {
if (is.func(middleware)) {
return function finalRunEffect(effect, effectId, currCb) {
const plainRunEffect = eff => runEffect(eff, effectId, currCb)
return middleware(plainRunEffect)(effect)
}
} else {
return runEffect
}
}
const env = {
stdChannel: channel,
dispatch: wrapSagaDispatch(dispatch),
getState,
sagaMonitor,
logError,
onError,
finalizeRunEffect,
}
// 新建task，作用是控制 Generator 流程，類似與自動流程管理，這個後面會講到
const task = proc(env, iterator, context, effectId, getMetaInfo(saga), null)
return task
}
複製程式碼
redux-saga的核心就是task, 控制generator函式saga執行流程。是一個複雜的自動流程管理，我們先看一個簡單的自動流程管理
// 一個返回promise的delay函式
const delay = (ms) => {
return new Promise((res) => {
setTimeout(res, ms);
});
}
function *main() {
yield delay(1000);
console.log('1s later');
yield delay(2000);
console.log('done');
}
// 為了達到想要的執行結果，我們必須在promise resolved之後再執行next statement，比如這樣
const gen = main();
const r1 = gen.next();
r1.value.then(() => {
const r2 = gen.next();
r2.value.then(() => {
gen.next();
})
})
複製程式碼
使用遞迴實現，自動流程控制
function autoRun(gfunc) {
const gen = gfunc();
function next() {
const res = gen.next();
if (res.done) return;
res.value.then(next);
}
next();
}
autoRun(main);
複製程式碼
上面的自動流程控制函式僅僅支援 promise。
proc
export default function proc(env, iterator, parentContext, parentEffectId, meta, cont) {
// ...
const task = newTask(parentEffectId, meta, cont)
const mainTask = { meta, cancel: cancelMain, _isRunning: true, _isCancelled: false }
// 構建 task tree
const taskQueue = forkQueue(
mainTask,
function onAbort() {
cancelledDueToErrorTasks.push(...taskQueue.getTaskNames())
},
end,
)
next()
// then return the task descriptor to the caller
return task
function next(arg, isErr) {
let result
if (isErr) {
result = iterator.throw(arg)
} else if (shouldCancel(arg)) {
// ...
} else if (shouldTerminate(arg)) {
// ...
} else {
result = iterator.next(arg)
}
if (!result.done) {
// 如果沒結束, 執行相應 effect
digestEffect(result.value, parentEffectId, '', next)
} else {
/**
This Generator has ended, terminate the main task and notify the fork queue
**/
mainTask._isRunning = false
mainTask.cont(result.value)
}
}
function digestEffect(effect, parentEffectId, label = '', cb) {
// 封裝了cb函式 增加了事件鉤子
function currCb(res, isErr) {
if (effectSettled) {
return
}
effectSettled = true
cb.cancel = noop // defensive measure
if (env.sagaMonitor) {
if (isErr) {
env.sagaMonitor.effectRejected(effectId, res)
} else {
env.sagaMonitor.effectResolved(effectId, res)
}
}
if (isErr) {
crashedEffect = effect
}
cb(res, isErr)
}
runEffect(effect, effectId, currCb)
}
// 每個 effect 的執行函式 這裡先看一下常用的幾個effect
function runEffect(effect, effectId, currCb) {
if (is.promise(effect)) {
resolvePromise(effect, currCb)
} else if (is.iterator(effect)) {
resolveIterator(effect, effectId, meta, currCb)
} else if (effect && effect[IO]) {
const { type, payload } = effect
if (type === effectTypes.TAKE) runTakeEffect(payload, currCb)
else if (type === effectTypes.PUT) runPutEffect(payload, currCb)
else if (type === effectTypes.CALL) runCallEffect(payload, effectId, currCb)
// 其他所有的effect ...
else currCb(effect)
} else {
// anything else returned as is
currCb(effect)
}
}
// 當返回值是 promise 時，就和之前實現的自動程序控制函式一樣嘛
function resolvePromise(promise, cb) {
// ...
promise.then(cb, error => cb(error, true))
}
// 當是generator函式時
function resolveIterator(iterator, effectId, meta, cb) {
proc(env, iterator, taskContext, effectId, meta, cb)
}
// 當是 take 就把callback放在channel裡，如果有匹配事件發生，觸發 callback
function runTakeEffect({ channel = env.stdChannel, pattern, maybe }, cb) {
const takeCb = input => {
if (input instanceof Error) {
cb(input, true)
return
}
if (isEnd(input) && !maybe) {
cb(TERMINATE)
return
}
cb(input)
}
try {
channel.take(takeCb, is.notUndef(pattern) ? matcher(pattern) : null)
} catch (err) {
cb(err, true)
return
}
cb.cancel = takeCb.cancel
}
function runPutEffect({ channel, action, resolve }, cb) {
asap(() => {
let result
try {
// 傳送 action
result = (channel ? channel.put : env.dispatch)(action)
} catch (error) {
cb(error, true)
return
}
if (resolve && is.promise(result)) {
resolvePromise(result, cb)
} else {
cb(result)
}
})
// put 是不能取消的
}
function runCallEffect({ context, fn, args }, effectId, cb) {
let result
try {
result = fn.apply(context, args)
} catch (error) {
cb(error, true)
return
}
return is.promise(result)
? resolvePromise(result, cb)
: is.iterator(result)
? resolveIterator(result, effectId, getMetaInfo(fn), cb)
: cb(result)
}
}
複製程式碼
總結
redux-saga 將非同步操作抽象為 effect，利用 generator 函式，控制saga流程。 到目前為止，只是涉及了一些基本流程，下一篇會對本篇進行補充。